kaynine-runtime 0.1.0

//! start_run flow and the authoritative-event persistence bridge.
//!
//! Design note (SPEC §5.3): the run task does NOT broadcast committed
//! authoritative events on the session feed. The loop's realtime events
//! (TurnStarted, deltas, TurnCompleted, RunCompleted, ...) already cover the
//! same information, so broadcasting both would double-emit. Instead the
//! forwarder stamps each realtime envelope with the latest committed revision
//! at forward time ("实时事件的 revision 取发出时最近一次已提交的
//! Revision"). `RunHooksImpl` below is persistence-only.

use crate::actor::{ActiveRun, ActorState};
use crate::approval::{ApprovalShared, InteractiveApprovalHandler};
use crate::service::{
    ApprovalOutcome, ApprovalResolution, CancelOutcome, CancelRequest, RunAccepted,
    SessionSnapshot, StartRunRequest, SteerAccepted, SteerRequest, UpdateSessionRequest,
};
use async_trait::async_trait;
use kaynine_core::agent_loop::{run_agent_loop, LoopOutcome, LoopParams, RunHooks};
use kaynine_core::compaction::{latest_valid_summary, CurrentModelSelector, SummaryPayload};
use kaynine_core::error::{KaynineError, RunFailureReason};
use kaynine_core::event::EventEnvelope;
use kaynine_core::ids::{BranchId, EntryId, ModelId, RunId, SessionId};
use kaynine_core::message::{ContentBlock, FinishReason, Message, ToolResultPayload};
use kaynine_core::policy::SteerSource;
use kaynine_core::provider::{GenerationOptions, ReasoningLevel};
use kaynine_core::store::{
    AppendOutcome, AuthoritativeEvent, CommandOutcome, CommandRecord, EntryRecord, LeaseOwner,
    RunRecord, RunState, SessionRecord, SessionStore, SteerRecord, SteerState, SummaryRecord,
    ToolCallRecord, ToolCallState,
};
use std::collections::{HashMap, VecDeque};
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::{broadcast, mpsc, oneshot, Mutex as AsyncMutex};
use tokio_util::sync::CancellationToken;

/// Grace the agent loop gives an executing tool after cancellation before it
/// synthesizes an in-doubt result. Shared between `LoopParams` and the
/// `ActiveRun` entry so the cancel handler knows exactly how long the run
/// task can take to append its terminal.
const CANCEL_GRACE: Duration = Duration::from_secs(30);
/// Extra slack on top of `CANCEL_GRACE` for terminal append + join after the
/// loop itself has returned.
const CANCEL_SLACK: Duration = Duration::from_secs(5);
/// How long the cancel handler polls the store for the committed terminal
/// after the run task joins (covers the append-vs-read visibility race).
const TERMINAL_POLL: Duration = Duration::from_secs(1);
const TERMINAL_POLL_INTERVAL: Duration = Duration::from_millis(50);

/// Steer queue shared by the actor (producer), the loop's `SteerSource`
/// (reader), and `RunHooksImpl::on_steer_applied` (drainer). `pending()` does
/// NOT remove items: only the runtime hook removes them when it persists
/// `SteerApplied`, so a steer is injected exactly once. The actor and run task
/// never race on it (single actor thread + the run task between safe points).
pub(crate) struct SteerQueue {
    items: Mutex<VecDeque<(String, String)>>,
}

impl SteerQueue {
    pub(crate) fn new() -> Arc<Self> {
        Arc::new(Self {
            items: Mutex::new(VecDeque::new()),
        })
    }

    pub(crate) fn push(&self, steer_id: String, content: String) {
        self.items
            .lock()
            .expect("steer queue mutex poisoned")
            .push_back((steer_id, content));
    }

    fn contents(&self) -> Vec<String> {
        self.items
            .lock()
            .expect("steer queue mutex poisoned")
            .iter()
            .map(|(_, content)| content.clone())
            .collect()
    }

    /// Removes the first `count` queued items, returning their
    /// (steer_id, content) pairs for SteerApplied persistence.
    fn take(&self, count: usize) -> Vec<(String, String)> {
        let mut items = self.items.lock().expect("steer queue mutex poisoned");
        (0..count.min(items.len()))
            .filter_map(|_| items.pop_front())
            .collect()
    }
}

#[async_trait]
impl SteerSource for SteerQueue {
    async fn pending(&self) -> Vec<String> {
        self.contents()
    }
}

pub(crate) async fn build_snapshot(
    store: &dyn SessionStore,
    session_id: &SessionId,
    active_run: Option<crate::service::ActiveRunInfo>,
    current_revision_override: Option<u64>,
    last_run_seq: Option<u64>,
) -> Result<SessionSnapshot, KaynineError> {
    let session = store
        .get_session(session_id)
        .await?
        .ok_or(KaynineError::SessionNotFound)?;
    let branches = store.list_branches(session_id).await?;
    let mut chains = HashMap::new();
    for branch in &branches {
        if let Some(head) = &branch.head_entry_id {
            chains.insert(branch.branch_id.clone(), store.load_chain(head).await?);
        }
    }
    let unapplied_steers = store
        .list_steers(session_id)
        .await?
        .into_iter()
        .filter(|steer| steer.state == SteerState::Unapplied)
        .collect();
    Ok(SessionSnapshot {
        current_revision: current_revision_override.unwrap_or(session.current_revision),
        session,
        branches,
        chains,
        active_run,
        last_run_seq,
        unapplied_steers,
    })
}

pub(crate) async fn handle_start_run(
    state: &mut ActorState,
    request: Box<StartRunRequest>,
    reply: oneshot::Sender<Result<RunAccepted, KaynineError>>,
) {
    let _ = reply.send(start_run(state, request).await);
}

async fn start_run(
    state: &mut ActorState,
    request: Box<StartRunRequest>,
) -> Result<RunAccepted, KaynineError> {
    if state.shutting {
        return Err(KaynineError::InvalidRequest);
    }
    if state.active.as_ref().is_some_and(|a| !a.join.is_finished()) {
        return Err(KaynineError::RunAlreadyActive);
    }
    state.active = None;

    // v0.1 start_run carries text only; image blobs need the blob-upload API
    // from a later milestone (ContentBlock::Image has a BinaryRef without
    // bytes, so there is nothing to persist into the blobs table here).
    if request
        .content
        .iter()
        .any(|block| matches!(block, ContentBlock::Image { .. }))
    {
        return Err(KaynineError::InvalidRequest);
    }

    let command = CommandRecord {
        command_id: request.command_id.clone(),
        kind: "start_run".to_string(),
        request_hash: start_run_hash(&request),
        result: None,
    };
    match state.store.record_command(command.clone()).await? {
        CommandOutcome::Accepted => {}
        CommandOutcome::Duplicate(record) => {
            return match record.result {
                Some(result) => serde_json::from_value::<RunAccepted>(result)
                    .map_err(|_| KaynineError::Internal),
                // A prior attempt crashed before completing; the caller must
                // retry with a new command id after inspecting state.
                None => Err(KaynineError::Internal),
            };
        }
        CommandOutcome::Conflict => return Err(KaynineError::IdempotencyConflict),
    }

    let session = state
        .store
        .get_session(&request.session_id)
        .await?
        .ok_or(KaynineError::SessionNotFound)?;
    let model = request
        .model_override
        .clone()
        .unwrap_or(session.default_model);
    let reasoning = request.reasoning_override.unwrap_or(ReasoningLevel::Off);

    let branches = state.store.list_branches(&request.session_id).await?;
    let branch = branches
        .iter()
        .find(|b| b.branch_id == request.branch_id)
        .ok_or(KaynineError::InvalidRequest)?;

    let run_id = RunId::generate();
    let request_branch = request.branch_id.clone();
    let user_entry = EntryRecord {
        entry_id: EntryId::generate(),
        parent_entry_id: branch.head_entry_id.clone(),
        branch_id: request.branch_id.clone(),
        message: Message::User {
            blocks: request.content.clone(),
        },
    };
    let run = RunRecord {
        run_id: run_id.clone(),
        session_id: request.session_id.clone(),
        branch_id: request.branch_id.clone(),
        state: RunState::Running,
        model: model.clone(),
        reasoning,
        failure: None,
        // 0 tells the store to stamp its own clock (sqlite unixepoch()).
        started_at_unix: 0,
    };

    let expected = *state.revision.lock().await;
    let outcome = state
        .store
        .append_events(
            &request.session_id,
            expected,
            &state.owner,
            vec![
                AuthoritativeEvent::UserMessageAccepted {
                    entry: user_entry.clone(),
                },
                AuthoritativeEvent::RunStarted { run },
            ],
            Vec::new(),
        )
        .await;

    let new_revision = match outcome {
        Ok(AppendOutcome::Appended { new_revision }) => new_revision,
        Ok(AppendOutcome::RevisionConflict { current_revision }) => {
            *state.revision.lock().await = current_revision;
            return Err(KaynineError::RevisionConflict);
        }
        Ok(_) => {
            tracing::error!(session_id = %request.session_id, "start_run append lost lease");
            return Err(KaynineError::Internal);
        }
        Err(error) => return Err(error),
    };
    *state.revision.lock().await = new_revision;

    let accepted = RunAccepted {
        run_id: run_id.clone(),
        user_entry_id: user_entry.entry_id.clone(),
        revision: new_revision,
    };
    if let Err(error) = state
        .store
        .complete_command(
            &command.command_id,
            serde_json::to_value(&accepted).map_err(|_| KaynineError::Internal)?,
        )
        .await
    {
        // The run is committed; a failed command-result write only costs a
        // retry-cycle Inconsistency, not the run itself.
        tracing::error!(?error, command_id = %command.command_id, "start_run command result write failed");
    }

    let definition_id = session.definition_id.clone();
    let cancel = CancellationToken::new();
    let steer_queue = SteerQueue::new();
    let approval_shared = Arc::new(ApprovalShared::default());
    let join = spawn_run_task(
        state.store.clone(),
        state.owner.clone(),
        state.revision.clone(),
        state.event_tx.clone(),
        state.run_seq.clone(),
        request,
        run_id.clone(),
        model.clone(),
        reasoning,
        user_entry.entry_id.clone(),
        definition_id,
        steer_queue.clone(),
        approval_shared.clone(),
        cancel.clone(),
    );
    state.active = Some(crate::actor::ActiveRun {
        run_id,
        branch_id: request_branch,
        model,
        cancel,
        join,
        cancel_grace: CANCEL_GRACE,
        steer_queue,
        approval_shared,
    });
    Ok(accepted)
}

pub(crate) async fn handle_steer(
    state: &mut ActorState,
    request: SteerRequest,
    reply: oneshot::Sender<Result<SteerAccepted, KaynineError>>,
) {
    let _ = reply.send(steer(state, request).await);
}

async fn steer(
    state: &mut ActorState,
    request: SteerRequest,
) -> Result<SteerAccepted, KaynineError> {
    // Documented v0.1 simplification: steering requires a live run; there is
    // no RunNotActive error variant, so map to InvalidRequest.
    let active = state
        .active
        .as_mut()
        .filter(|a| a.run_id == request.run_id && !a.join.is_finished())
        .ok_or(KaynineError::InvalidRequest)?;

    let command = CommandRecord {
        command_id: request.command_id.clone(),
        kind: "steer".to_string(),
        request_hash: format!(
            "{}/{}/{}",
            request.session_id, request.run_id, request.content
        ),
        result: None,
    };
    match state.store.record_command(command.clone()).await? {
        CommandOutcome::Accepted => {}
        CommandOutcome::Duplicate(record) => {
            return match record.result {
                Some(result) => serde_json::from_value::<SteerAccepted>(result)
                    .map_err(|_| KaynineError::Internal),
                None => Err(KaynineError::Internal),
            };
        }
        CommandOutcome::Conflict => return Err(KaynineError::IdempotencyConflict),
    }

    let steer_id = uuid::Uuid::new_v4().to_string();
    // Steer 到达即持久化 (SPEC §7): the queued input is durable before it is
    // visible to the run, so a crash leaves a recoverable Unapplied steer.
    // The run task commits concurrently (single writer lease, but the actor
    // and run task share the revision counter), so retry on conflict.
    let mut new_revision = None;
    for _ in 0..8 {
        let expected = *state.revision.lock().await;
        match state
            .store
            .append_events(
                &request.session_id,
                expected,
                &state.owner,
                vec![AuthoritativeEvent::SteerQueued {
                    steer: SteerRecord {
                        steer_id: steer_id.clone(),
                        session_id: request.session_id.clone(),
                        content: request.content.clone(),
                        state: SteerState::Queued,
                        received_at_unix: std::time::SystemTime::now()
                            .duration_since(std::time::UNIX_EPOCH)
                            .map(|d| d.as_secs() as i64)
                            .unwrap_or(0),
                    },
                }],
                Vec::new(),
            )
            .await
        {
            Ok(AppendOutcome::Appended {
                new_revision: revision,
            }) => {
                new_revision = Some(revision);
                break;
            }
            Ok(AppendOutcome::RevisionConflict { current_revision }) => {
                *state.revision.lock().await = current_revision;
                continue;
            }
            Ok(_) => return Err(KaynineError::Internal),
            Err(error) => return Err(error),
        }
    }
    let new_revision = new_revision.ok_or(KaynineError::RevisionConflict)?;
    *state.revision.lock().await = new_revision;

    // Only now does the run see it: injected at the next safe point.
    active
        .steer_queue
        .push(steer_id.clone(), request.content.clone());

    let accepted = SteerAccepted {
        steer_id,
        revision: new_revision,
    };
    if let Err(error) = state
        .store
        .complete_command(
            &command.command_id,
            serde_json::to_value(&accepted).map_err(|_| KaynineError::Internal)?,
        )
        .await
    {
        tracing::error!(?error, command_id = %command.command_id, "steer command result write failed");
    }
    Ok(accepted)
}

pub(crate) async fn handle_resolve_approval(
    state: &mut ActorState,
    request: ApprovalResolution,
    reply: oneshot::Sender<Result<ApprovalOutcome, KaynineError>>,
) {
    let _ = reply.send(resolve_approval(state, request).await);
}

async fn resolve_approval(
    state: &mut ActorState,
    request: ApprovalResolution,
) -> Result<ApprovalOutcome, KaynineError> {
    let command = CommandRecord {
        command_id: request.command_id.clone(),
        kind: "resolve_approval".to_string(),
        request_hash: format!(
            "{}/{}/{}/{}",
            request.session_id, request.run_id, request.call_id, request.approved
        ),
        result: None,
    };
    match state.store.record_command(command.clone()).await? {
        CommandOutcome::Accepted => {}
        CommandOutcome::Duplicate(record) => {
            return match record.result {
                Some(result) => serde_json::from_value::<ApprovalOutcome>(result)
                    .map_err(|_| KaynineError::Internal),
                None => Err(KaynineError::Internal),
            };
        }
        CommandOutcome::Conflict => return Err(KaynineError::IdempotencyConflict),
    }

    // Unlike steer, a resolution may legitimately arrive after the run task
    // finished (e.g. a late decision after the approval timed out); the
    // shared waiter state outlives the join until the actor drops the run.
    let outcome = match state.active.as_mut().filter(|a| a.run_id == request.run_id) {
        Some(active) => InteractiveApprovalHandler::resolve(
            &active.approval_shared,
            &request.call_id,
            request.approved,
        )
        .into(),
        None => ApprovalOutcome::NotFound,
    };

    if let Err(error) = state
        .store
        .complete_command(
            &command.command_id,
            serde_json::to_value(outcome).map_err(|_| KaynineError::Internal)?,
        )
        .await
    {
        tracing::error!(?error, command_id = %command.command_id, "resolve_approval command result write failed");
    }
    Ok(outcome)
}

pub(crate) async fn handle_cancel(
    state: &mut ActorState,
    request: CancelRequest,
    reply: oneshot::Sender<Result<CancelOutcome, KaynineError>>,
) {
    let _ = reply.send(cancel_run(state, request).await);
}

async fn cancel_run(
    state: &mut ActorState,
    request: CancelRequest,
) -> Result<CancelOutcome, KaynineError> {
    let command = CommandRecord {
        command_id: request.command_id.clone(),
        kind: "cancel".to_string(),
        request_hash: format!("{}/{}", request.session_id, request.run_id),
        result: None,
    };
    match state.store.record_command(command.clone()).await? {
        CommandOutcome::Accepted => {}
        CommandOutcome::Duplicate(record) => {
            return match record.result {
                Some(result) => serde_json::from_value::<CancelOutcome>(result)
                    .map_err(|_| KaynineError::Internal),
                None => Err(KaynineError::Internal),
            };
        }
        CommandOutcome::Conflict => return Err(KaynineError::IdempotencyConflict),
    }

    // Some(true): run was active and joined within grace.
    // Some(false): run was active but outlived grace + slack — impossible by
    // construction; fail loudly below without clearing active (no detached
    // concurrent writer). None: no live run with that id.
    let stopped = match state
        .active
        .as_mut()
        .filter(|a| a.run_id == request.run_id && !a.join.is_finished())
    {
        Some(active) => {
            active.cancel.cancel();
            // Invariant: after cancellation the loop terminates within
            // cancel_grace + epsilon (stream select cancels immediately, an
            // executing tool is bounded by cancel_grace, remaining calls are
            // synthesized), so grace + slack always suffices.
            match tokio::time::timeout(active.cancel_grace + CANCEL_SLACK, &mut active.join).await {
                Ok(_) => Some(true),
                Err(_) => Some(false),
            }
        }
        None => None,
    };

    let outcome = match stopped {
        Some(false) => {
            // Keep `active` set: the run task is still appending. Clearing it
            // here would let a new run start and race the detached writer.
            tracing::error!(
                run_id = %request.run_id,
                "run failed to stop within cancel grace; refusing to detach"
            );
            return Err(KaynineError::Internal);
        }
        None => {
            state.active = None;
            let run = state
                .store
                .get_run(&request.run_id)
                .await?
                .ok_or(KaynineError::InvalidRequest)?;
            CancelOutcome::AlreadyTerminal { state: run.state }
        }
        Some(true) => {
            state.active = None;
            // The run task has joined; its terminal append is the last write.
            // Poll briefly to cover the join-vs-store-visibility race.
            let deadline = tokio::time::Instant::now() + TERMINAL_POLL;
            let run_state = loop {
                let run = state
                    .store
                    .get_run(&request.run_id)
                    .await?
                    .ok_or(KaynineError::InvalidRequest)?;
                if run.state != RunState::Running || tokio::time::Instant::now() >= deadline {
                    break run.state;
                }
                tokio::time::sleep(TERMINAL_POLL_INTERVAL).await;
            };
            match run_state {
                RunState::Cancelled => CancelOutcome::Cancelled {
                    revision: *state.revision.lock().await,
                },
                RunState::Running => {
                    tracing::error!(
                        run_id = %request.run_id,
                        "run joined but no terminal committed after polling"
                    );
                    return Err(KaynineError::Internal);
                }
                // Completed/Failed/Interrupted: the run finished on its own
                // before the cancel landed; report the committed truth.
                committed => CancelOutcome::AlreadyTerminal { state: committed },
            }
        }
    };
    if let Err(error) = state
        .store
        .complete_command(
            &command.command_id,
            serde_json::to_value(&outcome).map_err(|_| KaynineError::Internal)?,
        )
        .await
    {
        tracing::error!(?error, command_id = %command.command_id, "cancel command result write failed");
    }
    Ok(outcome)
}

fn start_run_hash(request: &StartRunRequest) -> String {
    serde_json::json!({
        "session_id": request.session_id,
        "branch_id": request.branch_id,
        "content": request.content,
        "model_override": request.model_override,
        "reasoning_override": request.reasoning_override,
    })
    .to_string()
}

pub(crate) async fn handle_update_session(
    state: &mut ActorState,
    request: UpdateSessionRequest,
    reply: oneshot::Sender<Result<SessionRecord, KaynineError>>,
) {
    let _ = reply.send(update_session(state, request).await);
}

async fn update_session(
    state: &mut ActorState,
    request: UpdateSessionRequest,
) -> Result<SessionRecord, KaynineError> {
    if state.active.as_ref().is_some_and(|a| !a.join.is_finished()) {
        return Err(KaynineError::SessionBusy);
    }
    state.active = None;

    let command = match &request.command_id {
        Some(command_id) => {
            let record = CommandRecord {
                command_id: command_id.clone(),
                kind: "update_session".to_string(),
                request_hash: serde_json::json!({
                    "session_id": request.session_id,
                    "default_model": request.default_model,
                    "reasoning": request.reasoning,
                    "metadata": request.metadata,
                })
                .to_string(),
                result: None,
            };
            match state.store.record_command(record.clone()).await? {
                CommandOutcome::Accepted => Some(record),
                CommandOutcome::Duplicate(record) => {
                    return match record.result {
                        Some(result) => serde_json::from_value::<SessionRecord>(result)
                            .map_err(|_| KaynineError::Internal),
                        None => Err(KaynineError::Internal),
                    };
                }
                CommandOutcome::Conflict => return Err(KaynineError::IdempotencyConflict),
            }
        }
        None => None,
    };

    let session = state
        .store
        .get_session(&request.session_id)
        .await?
        .ok_or(KaynineError::SessionNotFound)?;

    let mut events = Vec::new();
    if request.default_model.is_some() || request.reasoning.is_some() {
        events.push(AuthoritativeEvent::ModelChanged {
            model: request
                .default_model
                .clone()
                .unwrap_or_else(|| session.default_model.clone()),
            reasoning: request.reasoning.unwrap_or(ReasoningLevel::Off),
        });
    }
    if let Some(metadata) = request.metadata.clone() {
        events.push(AuthoritativeEvent::SessionMetadataUpdated { metadata });
    }
    if events.is_empty() {
        return Ok(session);
    }

    let expected = *state.revision.lock().await;
    let new_revision = match state
        .store
        .append_events(
            &request.session_id,
            expected,
            &state.owner,
            events,
            Vec::new(),
        )
        .await
    {
        Ok(AppendOutcome::Appended { new_revision }) => new_revision,
        Ok(AppendOutcome::RevisionConflict { current_revision }) => {
            *state.revision.lock().await = current_revision;
            return Err(KaynineError::RevisionConflict);
        }
        Ok(_) => return Err(KaynineError::Internal),
        Err(error) => return Err(error),
    };
    *state.revision.lock().await = new_revision;

    let refreshed = state
        .store
        .get_session(&request.session_id)
        .await?
        .ok_or(KaynineError::SessionNotFound)?;
    if let Some(command) = command {
        if let Err(error) = state
            .store
            .complete_command(
                &command.command_id,
                serde_json::to_value(&refreshed).map_err(|_| KaynineError::Internal)?,
            )
            .await
        {
            tracing::error!(?error, command_id = %command.command_id, "update_session command result write failed");
        }
    }
    Ok(refreshed)
}

/// Cancels the actor's active run and waits up to `grace` for it to append
/// its own terminal. Returns (cancelled, interrupted) run ids; interrupted
/// runs get an `Interrupted` terminal appended here because their task
/// outlived the grace period.
pub(crate) async fn handle_shutdown_actor(
    state: &mut ActorState,
    grace: Duration,
) -> (Vec<RunId>, Vec<RunId>) {
    let mut cancelled = Vec::new();
    let mut interrupted = Vec::new();
    if let Some(active) = state.active.take() {
        let ActiveRun {
            run_id,
            cancel,
            join,
            ..
        } = active;
        if !join.is_finished() {
            cancel.cancel();
            let mut join = Box::pin(join);
            match tokio::time::timeout(grace, join.as_mut()).await {
                Ok(_) => cancelled.push(run_id.clone()),
                Err(_) => {
                    append_terminal(
                        &state.store,
                        &state.session_id,
                        &state.owner,
                        &state.revision,
                        &run_id,
                        RunState::Interrupted,
                        None,
                    )
                    .await;
                    // The detached loop task must not append a competing
                    // terminal after our Interrupted write.
                    join.abort();
                    interrupted.push(run_id);
                }
            }
        }
    }
    (cancelled, interrupted)
}

#[allow(clippy::too_many_arguments)]
fn spawn_run_task(
    store: Arc<dyn SessionStore>,
    owner: LeaseOwner,
    revision: Arc<AsyncMutex<u64>>,
    event_tx: broadcast::Sender<EventEnvelope<kaynine_core::event::RealtimeEvent>>,
    run_seq: Arc<AtomicU64>,
    request: Box<StartRunRequest>,
    run_id: RunId,
    model: ModelId,
    reasoning: ReasoningLevel,
    user_entry_id: EntryId,
    definition_id: String,
    steer_queue: Arc<SteerQueue>,
    approval_shared: Arc<ApprovalShared>,
    cancel: CancellationToken,
) -> tokio::task::JoinHandle<()> {
    let session_id = request.session_id.clone();
    let branch_id = request.branch_id.clone();
    let provider_id = request.provider.id();
    tokio::spawn(async move {
        // Working context = loaded chain with the latest still-valid summary
        // checkpoint applied over its covered prefix (SPEC §10 lazy
        // restore). Original entries are never deleted; a fork whose chain
        // no longer contains covered_until_entry simply skips the restore.
        let (history, entry_tail): (Vec<Message>, Vec<EntryId>) =
            match store.load_chain(&user_entry_id).await {
                Ok(entries) => {
                    let entry_ids: Vec<EntryId> =
                        entries.iter().map(|e| e.entry_id.clone()).collect();
                    let messages: Vec<Message> = entries.into_iter().map(|e| e.message).collect();
                    match store.list_summaries(&branch_id).await {
                        Ok(summaries) => match latest_valid_summary(&entry_ids, &summaries) {
                            Some((position, record)) => {
                                let mut working = Vec::with_capacity(
                                    1 + entry_ids.len().saturating_sub(position + 1),
                                );
                                working.push(Message::Summary {
                                    text: record.text.clone(),
                                });
                                working.extend(messages.iter().skip(position + 1).cloned());
                                // The placeholder id keeps the message↔entry
                                // mapping aligned for the next checkpoint.
                                let mut tail = Vec::with_capacity(entry_ids.len() - position);
                                tail.push(record.covered_until_entry.clone());
                                tail.extend(entry_ids.iter().skip(position + 1).cloned());
                                (working, tail)
                            }
                            None => (messages, entry_ids),
                        },
                        Err(error) => {
                            tracing::warn!(
                                ?error,
                                "list_summaries failed; running without summary restore"
                            );
                            (messages, entry_ids)
                        }
                    }
                }
                Err(error) => {
                    tracing::error!(
                        ?error,
                        "run failed to load chain; appending failed terminal"
                    );
                    append_terminal(
                        &store,
                        &session_id,
                        &owner,
                        &revision,
                        &run_id,
                        RunState::Failed,
                        Some(RunFailureReason::Internal),
                    )
                    .await;
                    return;
                }
            };

        let tail: Arc<std::sync::Mutex<EntryId>> =
            Arc::new(std::sync::Mutex::new(user_entry_id.clone()));
        let hooks = Arc::new(RunHooksImpl {
            store: store.clone(),
            session_id: session_id.clone(),
            branch_id: branch_id.clone(),
            run_id: run_id.clone(),
            owner: owner.clone(),
            revision: revision.clone(),
            tail: tail.clone(),
            // Entry ids aligned 1:1 with the loop's working messages at run
            // start (see the restore above); entries appended by the hooks
            // below keep the alignment (tool-result batches are 1:1 per
            // result entry — a documented v0.1 approximation).
            entry_tail: Arc::new(std::sync::Mutex::new(entry_tail)),
            turn: AtomicU32::new(0),
            steer_queue: steer_queue.clone(),
            provider: provider_id,
            model: model.clone(),
        });

        let (events_tx, mut events_rx) =
            mpsc::channel::<EventEnvelope<kaynine_core::event::RealtimeEvent>>(256);
        // The interactive approval handler shares the loop's realtime channel
        // so its events reach the session feed through the same forwarder.
        let approval = request.approval_timeout.map(|timeout| {
            Arc::new(InteractiveApprovalHandler::new(
                store.clone(),
                session_id.clone(),
                branch_id.clone(),
                run_id.clone(),
                revision.clone(),
                owner.clone(),
                events_tx.clone(),
                approval_shared,
                timeout,
            )) as Arc<dyn kaynine_core::policy::ApprovalHandler>
        });
        let params = LoopParams {
            session_id: session_id.clone(),
            branch_id: branch_id.clone(),
            run_id: run_id.clone(),
            provider: request.provider.clone(),
            token_counter: request.token_counter.clone(),
            credentials: request.credentials.clone(),
            tools: request.tools.clone(),
            model,
            reasoning,
            generation: GenerationOptions::default(),
            capabilities: request.capabilities,
            system_prompt: request.system_prompt.clone(),
            prompt: request.prompt.clone(),
            skills: None,
            definition_id,
            history,
            compaction: request.compaction,
            compaction_selector: request
                .compaction_selector
                .unwrap_or_else(|| Arc::new(CurrentModelSelector)),
            provider_options: serde_json::Value::Null,
            budget: request.budget,
            events: events_tx,
            cancel: cancel.clone(),
            max_turns: request.max_turns,
            hooks: hooks.clone(),
            cancel_grace: CANCEL_GRACE,
            policy: request.policy.clone(),
            approval,
            steer: Some(steer_queue),
        };

        let loop_join = tokio::spawn(run_agent_loop(params));
        while let Some(mut envelope) = events_rx.recv().await {
            // Realtime events carry the latest committed revision at forward
            // time (SPEC §5.3); they never advance the revision themselves.
            envelope.revision = *revision.lock().await;
            match envelope.run_seq {
                Some(seq) => {
                    run_seq.fetch_max(seq, Ordering::SeqCst);
                }
                // Convention: envelopes arriving with run_seq None come from
                // the runtime (approval handler), not the loop's dense 1..n
                // counter; the forwarder assigns the next value from the
                // shared high-water counter. Approval events only flow while
                // the loop is blocked awaiting the decision, so the assigned
                // value may duplicate the loop's next seq number — consumers
                // dedupe by (revision, run_seq).
                None => {
                    envelope.run_seq = Some(run_seq.fetch_add(1, Ordering::SeqCst) + 1);
                }
            }
            let _ = event_tx.send(envelope);
        }
        let outcome = match loop_join.await {
            Ok(outcome) => outcome,
            Err(join_error) => {
                tracing::error!(?join_error, "agent loop task panicked");
                LoopOutcome::Failed {
                    reason: RunFailureReason::Internal,
                    messages: Vec::new(),
                }
            }
        };

        let (terminal_state, failure) = match outcome {
            LoopOutcome::Completed { .. } => (RunState::Completed, None),
            LoopOutcome::Cancelled { .. } => (RunState::Cancelled, None),
            LoopOutcome::Failed { reason, .. } => (RunState::Failed, Some(reason)),
        };
        append_terminal(
            &store,
            &session_id,
            &owner,
            &revision,
            &run_id,
            terminal_state,
            failure,
        )
        .await;
    })
}

pub(crate) async fn append_terminal(
    store: &Arc<dyn SessionStore>,
    session_id: &SessionId,
    owner: &LeaseOwner,
    revision: &Arc<AsyncMutex<u64>>,
    run_id: &RunId,
    state: RunState,
    failure: Option<RunFailureReason>,
) {
    let expected = *revision.lock().await;
    match store
        .append_events(
            session_id,
            expected,
            owner,
            vec![AuthoritativeEvent::RunTerminal {
                run_id: run_id.clone(),
                state,
                failure,
            }],
            Vec::new(),
        )
        .await
    {
        Ok(AppendOutcome::Appended { new_revision }) => {
            *revision.lock().await = new_revision;
        }
        other => {
            // Single writer: a conflict here means something is deeply wrong.
            tracing::error!(?other, ?state, run_id = %run_id, "run terminal append failed");
        }
    }
}

/// Persistence-only bridge: every hook commits its authoritative event(s)
/// before the loop proceeds (SPEC §5.4 "先落库再执行"). It never broadcasts;
/// the run task's forwarder is the only thing on the session feed.
struct RunHooksImpl {
    store: Arc<dyn SessionStore>,
    session_id: SessionId,
    branch_id: BranchId,
    run_id: RunId,
    owner: LeaseOwner,
    revision: Arc<AsyncMutex<u64>>,
    /// Chain tail the next entry will hang off; updated after every commit.
    tail: Arc<std::sync::Mutex<EntryId>>,
    /// Entry ids aligned with the loop's working message list; used to map
    /// `SummaryPayload::covered_message_count` to a covered_until_entry.
    entry_tail: Arc<std::sync::Mutex<Vec<EntryId>>>,
    turn: AtomicU32,
    /// Provider/model recorded on persisted summary checkpoints.
    provider: kaynine_core::ids::ProviderId,
    model: ModelId,
    /// Shared with the actor (producer) and the loop's SteerSource; drained
    /// here so every injected steer is persisted exactly once.
    steer_queue: Arc<SteerQueue>,
}

impl RunHooksImpl {
    async fn append(&self, events: Vec<AuthoritativeEvent>) -> Result<(), RunFailureReason> {
        if events.is_empty() {
            return Ok(());
        }
        // The actor may commit concurrently through the shared revision
        // counter (e.g. SteerQueued), so retry on conflict like steer().
        for _ in 0..8 {
            let expected = *self.revision.lock().await;
            match self
                .store
                .append_events(
                    &self.session_id,
                    expected,
                    &self.owner,
                    events.clone(),
                    Vec::new(),
                )
                .await
            {
                Ok(AppendOutcome::Appended { new_revision }) => {
                    *self.revision.lock().await = new_revision;
                    return Ok(());
                }
                Ok(AppendOutcome::RevisionConflict { current_revision }) => {
                    *self.revision.lock().await = current_revision;
                    continue;
                }
                Ok(other) => {
                    tracing::error!(?other, session_id = %self.session_id, "run hook append lost lease");
                    return Err(RunFailureReason::Internal);
                }
                Err(error) => {
                    tracing::error!(?error, session_id = %self.session_id, "run hook append failed");
                    return Err(RunFailureReason::Internal);
                }
            }
        }
        Err(RunFailureReason::Internal)
    }

    fn take_tail(&self) -> EntryId {
        self.tail.lock().expect("tail mutex poisoned").clone()
    }

    fn set_tail(&self, entry_id: EntryId) {
        *self.tail.lock().expect("tail mutex poisoned") = entry_id;
    }
}

#[async_trait]
impl RunHooks for RunHooksImpl {
    async fn on_turn_started(&self, turn: u32) -> Result<(), RunFailureReason> {
        self.turn.store(turn, Ordering::SeqCst);
        self.append(vec![AuthoritativeEvent::TurnStarted {
            run_id: self.run_id.clone(),
            turn,
        }])
        .await
    }

    async fn on_assistant_message(&self, message: &Message) -> Result<(), RunFailureReason> {
        let entry = EntryRecord {
            entry_id: EntryId::generate(),
            parent_entry_id: Some(self.take_tail()),
            branch_id: self.branch_id.clone(),
            message: message.clone(),
        };
        let entry_id = entry.entry_id.clone();
        self.append(vec![AuthoritativeEvent::AssistantMessagePersisted {
            entry,
        }])
        .await?;
        self.set_tail(entry_id.clone());
        self.entry_tail
            .lock()
            .expect("entry tail mutex poisoned")
            .push(entry_id);
        Ok(())
    }

    async fn on_tool_calls_planned(
        &self,
        calls: &[(kaynine_core::ids::ToolCallId, String, serde_json::Value)],
    ) -> Result<(), RunFailureReason> {
        let turn = self.turn.load(Ordering::SeqCst);
        let events = calls
            .iter()
            .map(
                |(call_id, name, arguments)| AuthoritativeEvent::ToolCallPlanned {
                    call: ToolCallRecord {
                        call_id: call_id.clone(),
                        run_id: self.run_id.clone(),
                        turn,
                        name: name.clone(),
                        arguments: arguments.clone(),
                        state: ToolCallState::Planned,
                    },
                },
            )
            .collect();
        self.append(events).await
    }

    async fn on_tool_results(&self, results: &[ToolResultPayload]) -> Result<(), RunFailureReason> {
        // One entry per result keeps every ToolResultRecorded event pointing
        // at a distinct entry (the store applies each event's entry insert and
        // call-state update in the same transaction).
        let mut events = Vec::with_capacity(results.len());
        let mut tail = self.take_tail();
        let mut new_entry_ids = Vec::with_capacity(results.len());
        for result in results {
            let entry = EntryRecord {
                entry_id: EntryId::generate(),
                parent_entry_id: Some(tail.clone()),
                branch_id: self.branch_id.clone(),
                message: Message::ToolResult {
                    results: vec![result.clone()],
                },
            };
            tail = entry.entry_id.clone();
            new_entry_ids.push(entry.entry_id.clone());
            events.push(AuthoritativeEvent::ToolResultRecorded {
                entry,
                call_id: result.call_id.clone(),
                synthesized: result.is_error,
            });
        }
        self.append(events).await?;
        self.set_tail(tail);
        self.entry_tail
            .lock()
            .expect("entry tail mutex poisoned")
            .extend(new_entry_ids);
        Ok(())
    }

    async fn on_turn_completed(
        &self,
        turn: u32,
        finish_reason: FinishReason,
    ) -> Result<(), RunFailureReason> {
        self.append(vec![AuthoritativeEvent::TurnCompleted {
            run_id: self.run_id.clone(),
            turn,
            finish_reason,
        }])
        .await
    }

    /// Persists one SteerApplied + User entry per injected content, in arrival
    /// order, hanging each entry off the current chain tail (SPEC §7).
    async fn on_steer_applied(&self, contents: &[String]) -> Result<(), RunFailureReason> {
        let taken = self.steer_queue.take(contents.len());
        if taken.is_empty() {
            return Ok(());
        }
        let mut events = Vec::with_capacity(taken.len());
        let mut tail = self.take_tail();
        let mut new_entry_ids = Vec::with_capacity(taken.len());
        for (steer_id, content) in taken {
            let entry = EntryRecord {
                entry_id: EntryId::generate(),
                parent_entry_id: Some(tail.clone()),
                branch_id: self.branch_id.clone(),
                message: Message::User {
                    blocks: vec![ContentBlock::Text { text: content }],
                },
            };
            tail = entry.entry_id.clone();
            new_entry_ids.push(entry.entry_id.clone());
            events.push(AuthoritativeEvent::SteerApplied { steer_id, entry });
        }
        self.append(events).await?;
        self.set_tail(tail);
        self.entry_tail
            .lock()
            .expect("entry tail mutex poisoned")
            .extend(new_entry_ids);
        Ok(())
    }

    /// Persists the SummaryCheckpoint (SPEC §10) with retry-on-conflict (the
    /// actor may commit concurrently through the shared revision counter),
    /// then realigns the message→entry mapping with the loop's post-
    /// compaction working list.
    async fn on_summary_checkpoint(
        &self,
        summary: &SummaryPayload,
    ) -> Result<(), RunFailureReason> {
        let (covered_until, retained_entries) = {
            let tail = self.entry_tail.lock().expect("entry tail mutex poisoned");
            let covered_until = summary
                .covered_message_count
                .checked_sub(1)
                .and_then(|i| tail.get(i).cloned())
                .or_else(|| tail.last().cloned())
                .ok_or(RunFailureReason::Internal)?;
            let retained = tail
                .iter()
                .skip(summary.retain_from)
                .cloned()
                .collect::<Vec<_>>();
            (covered_until, retained)
        };
        let record = SummaryRecord {
            summary_id: uuid::Uuid::new_v4().to_string(),
            branch_id: self.branch_id.clone(),
            covered_until_entry: covered_until.clone(),
            text: summary.text.clone(),
            source_hash: summary.source_hash.clone(),
            provider: self.provider.clone(),
            model: self.model.clone(),
            prompt_version: String::new(),
            usage: summary.usage,
        };
        let mut appended = false;
        for _ in 0..8 {
            let expected = *self.revision.lock().await;
            match self
                .store
                .append_events(
                    &self.session_id,
                    expected,
                    &self.owner,
                    vec![AuthoritativeEvent::SummaryCheckpoint {
                        summary: record.clone(),
                    }],
                    Vec::new(),
                )
                .await
            {
                Ok(AppendOutcome::Appended { new_revision }) => {
                    *self.revision.lock().await = new_revision;
                    appended = true;
                    break;
                }
                Ok(AppendOutcome::RevisionConflict { current_revision }) => {
                    *self.revision.lock().await = current_revision;
                    continue;
                }
                Ok(_) => return Err(RunFailureReason::Internal),
                Err(error) => {
                    tracing::error!(?error, session_id = %self.session_id, "summary checkpoint append failed");
                    return Err(RunFailureReason::Internal);
                }
            }
        }
        if !appended {
            return Err(RunFailureReason::Internal);
        }
        // Realign: the loop's working list is now [Summary] + retained
        // messages; the checkpoint's covered entry stands in for the cover.
        let mut realigned = Vec::with_capacity(1 + retained_entries.len());
        realigned.push(covered_until);
        realigned.extend(retained_entries);
        *self.entry_tail.lock().expect("entry tail mutex poisoned") = realigned;
        Ok(())
    }
}