kcode_kennedy_sessions/

lib.rs

//! Kennedy's complete logical session lifecycle and agent orchestration.

#![forbid(unsafe_code)]

mod services;

pub use kcode_kennedy_session_objects::ResolvedObject;
pub use kcode_telegram_session_coordinator::validate_file_name as validate_delivery_file_name;
pub use services::{Api as Service, LocalServices as Capabilities};

use std::{
    collections::{BTreeMap, BTreeSet, HashMap},
    future::Future,
    sync::{Arc, Weak},
    time::{Duration, Instant},
};

use anyhow::Context as _;
use chrono::{DateTime, Utc};
use kcode_agent_runtime::SessionHost as _;
use kcode_commit_session::{CommitReceipt, CommitRequest};
use kcode_dev_tools::{
    ATTACH_OBJECT_WEB_LIB_TOOL, CALL_RUST_BIN_TOOL, RUST_BIN_TOOLS, RUST_LIB_TOOLS, WEB_LIB_TOOLS,
    proposed_write_snapshot,
};
use kcode_dev_tools_chatend::{
    FreeformWrite, SourceSnapshot, apply_snapshot, decode_freeform_write, prepare_freeform_write,
};
use kcode_history_ingress_context::{
    Outcome as HistoryIngressContextOutcome, RecoveryOutcome as ContextRecoveryOutcome,
};
use kcode_kennedy_kweb_loader::{load_durable_batch, node_from_value};
use kcode_kennedy_kweb_plan::{
    Mutation as KwebMutation, Plan as KwebPlan, referenced_pending_nodes,
};
use kcode_kennedy_session_ingress::{is_terminal_external_response, restore_pending_turn};
use kcode_kennedy_session_presentation::{RenderRequest, render};
use kcode_kennedy_session_tool_contracts::{
    DecodedTool, ManagedObjectArguments, ValidationRequest, decode, decode_managed_objects,
    decode_note_to_self, validate,
};
use kcode_kennedy_session_tool_presentation::invocation_box_content;
use kcode_kennedy_subagent_context::Context as SubagentContext;
use kcode_kweb_context::{Context as KwebContext, Node as KwebNode};
use kcode_kweb_db::NodeId;
use kcode_server_object_envelopes::encode_file;
use kcode_session_history::{
    ErrorKind as HistoryErrorKind, LaunchSession as HistoryLaunchSession, NewSession,
    Session as HistorySession,
    chatend::{
        BoxContent, BoxId, BoxOwner, CacheExpectation, ContextProjection, Event, EventId,
        EventKind, PreparedProviderResume, ProviderContext, ProviderToolDefinition, SessionKind,
        SessionMetadata,
    },
};
use kcode_session_runtime_budget::{RoundBudget, RuntimeBudget, TimeBudget, TimeBudgetKind};
use kcode_speaker_system::KTOOLS as SPEECH_CLASSIFICATION_TOOLS;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use uuid::Uuid;

const BROWSER_CONVERSATION_REQUEST_TIMEOUT: Duration = Duration::from_secs(225 * 60);
const HISTORY_INGRESS_REQUEST_TIMEOUT: Duration = Duration::from_secs(225 * 60);
const HISTORY_INGRESS_ATTEMPT_DURATION: Duration = Duration::from_secs(45 * 60);
const WAKEUP_REQUEST_TIMEOUT: Duration = Duration::from_secs(225 * 60);
const SELF_TIME_HARD_STOP_ALLOWANCE: Duration = Duration::from_secs(15 * 60);
const MAX_MEDIA_ENRICHMENT_BYTES: u64 = 20 * 1024 * 1024;
const MAX_LAUNCH_INTENTS_PER_USER_TURN: usize = 10;
const LAUNCH_SESSION_TOOL: &str = "LaunchSession";
const KWEB_TOOL_INSTANCE: &str = "kweb";
const TASK_BOARD_TOOLS: [&str; 8] = [
    "CreateTaskCategory",
    "GetTaskCategory",
    "RemoveTaskCategory",
    "CreateTask",
    "GetTask",
    "UpdateTask",
    "RemoveTask",
    "GetTopTaskOrphan",
];
const CONTEXT_OVERFLOW_WARNING_BOX_NAME: &str = "Context overflow warning";
const CONTEXT_OVERFLOW_WARNING: &str = "Context size was exceeded, some context has been dehydrated. The session is now at risk of destabilizing, please perform any cleanup tasks and end the session";
const INGRESS_FORCE_COMMIT_NOTE: &str = "ingress_force_commit";
const CHECKPOINT_STATE_VERSION: u64 = 5;

#[derive(Debug)]
struct IngressTimeExpired;

impl std::fmt::Display for IngressTimeExpired {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str("history ingress time expired before EndSession")
    }
}

impl std::error::Error for IngressTimeExpired {}

/// Returns whether an error represents expiry of the shared ingress attempt timer.
pub fn is_ingress_time_expired(error: &anyhow::Error) -> bool {
    error.is::<IngressTimeExpired>()
}

fn ingress_time_remaining_at(deadline: &mut Option<Instant>, now: Instant) -> anyhow::Result<u64> {
    let Some(current) = *deadline else {
        *deadline = Some(
            now.checked_add(HISTORY_INGRESS_ATTEMPT_DURATION)
                .context("history ingress deadline overflow")?,
        );
        return Ok(HISTORY_INGRESS_ATTEMPT_DURATION.as_secs());
    };
    if now >= current {
        return Err(anyhow::Error::new(IngressTimeExpired));
    }
    Ok(current.duration_since(now).as_secs())
}

/// Application-selected primary model facts used mechanically by a session.
#[derive(Clone, Debug)]
pub struct RuntimeModel {
    pub model: String,
    pub reasoning_effort: String,
    pub context_window_tokens: u64,
}

impl RuntimeModel {
    pub fn from_intelligence(runtime: kcode_intelligence_router::RuntimeModel) -> Self {
        Self {
            model: runtime.model,
            reasoning_effort: runtime.reasoning_effort,
            context_window_tokens: runtime.context_window_tokens,
        }
    }

    fn attribution(&self) -> String {
        format!("{}-{}", self.model, self.reasoning_effort)
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum AgentMode {
    Conversation,
    FreeTime,
    Wakeup,
    Ingress { record_id: Option<String> },
}

/// Kind of finite enclosing deadline supplied by the workflow owner.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TurnDeadlineKind {
    /// Hard deadline for an inbound Telegram event.
    Telegram,
    /// Bounded cleanup deadline after user-selected self time.
    SelfTimeHardStop,
}

/// Absolute enclosing deadline that remains transient during a turn.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct TurnDeadline {
    /// Semantic category rendered in the provider footer.
    pub kind: TurnDeadlineKind,
    /// Absolute UTC deadline owned and enforced by orchestration.
    pub at: DateTime<Utc>,
}

#[derive(Clone, Debug)]
pub struct SessionOptions {
    pub session_type: String,
    pub root_node_ids: Vec<String>,
    pub reference_root_node_ids: Vec<String>,
    pub channel: Value,
    pub free_time: Value,
    pub orchestration: Value,
    pub provenance_id: Option<String>,
    pub mode: AgentMode,
    pub source_session_type: Option<String>,
    pub group_context: Value,
    pub rust_lib_session_id: Option<String>,
}

impl SessionOptions {
    pub fn conversation(session_type: impl Into<String>, roots: Vec<String>) -> Self {
        Self {
            session_type: session_type.into(),
            root_node_ids: roots,
            reference_root_node_ids: Vec::new(),
            channel: Value::Null,
            free_time: Value::Null,
            orchestration: json!({"owner":"backend","status":"idle"}),
            provenance_id: None,
            mode: AgentMode::Conversation,
            source_session_type: None,
            group_context: Value::Null,
            rust_lib_session_id: None,
        }
    }
}

fn restore_session_type(options: &mut SessionOptions, state: &Value) {
    if !matches!(&options.mode, AgentMode::Ingress { .. }) {
        options.session_type = state
            .get("sessionType")
            .and_then(Value::as_str)
            .unwrap_or(&options.session_type)
            .to_owned();
    }
}

fn restore_commit_receipt(restored: Option<&Value>) -> anyhow::Result<Option<CommitReceipt>> {
    restored
        .and_then(|state| state.get("commitReceipt"))
        .filter(|receipt| !receipt.is_null())
        .cloned()
        .map(serde_json::from_value)
        .transpose()
        .context("decoding the stored session commit receipt")
}

fn journal_kweb_plan(journal: &HistorySession) -> Option<&Value> {
    journal
        .state()
        .current_ingress_attempt_events()
        .iter()
        .rev()
        .find_map(|event| {
            let EventKind::KwebPlanChanged { operation } = &event.kind else {
                return None;
            };
            operation.get("plan")
        })
}

#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct LaunchSessionArguments {
    directive: String,
    context_node_ids: Vec<String>,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct LaunchIntent {
    invocation_id: String,
    user_turn_id: EventId,
    started_at: String,
    parent_session_id: String,
    effective_context_tokens: u64,
    root_node_ids: Vec<String>,
    reference_root_node_ids: Vec<String>,
    context_node_ids: Vec<String>,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct LaunchSuccess<'a> {
    session_id: &'a str,
    command_id: &'a str,
}

fn decode_launch_session_arguments(value: &Value) -> anyhow::Result<LaunchSessionArguments> {
    let arguments: LaunchSessionArguments =
        serde_json::from_value(value.clone()).context("LaunchSession arguments are invalid")?;
    anyhow::ensure!(
        !arguments.directive.trim().is_empty(),
        "LaunchSession directive must not be blank"
    );
    validate_canonical_distinct_ids(&arguments.context_node_ids, "context node")?;
    Ok(arguments)
}

fn validate_canonical_distinct_ids(ids: &[String], label: &str) -> anyhow::Result<()> {
    let mut seen = BTreeSet::new();
    for id in ids {
        canonical_id(id).with_context(|| format!("LaunchSession {label} ID is invalid"))?;
        anyhow::ensure!(
            seen.insert(id.as_str()),
            "LaunchSession {label} ID {id} is duplicated"
        );
    }
    Ok(())
}

fn validate_loaded_launch_context(
    context: &KwebContext,
    context_node_ids: &[String],
) -> anyhow::Result<()> {
    for id in context_node_ids {
        anyhow::ensure!(
            context.contains_full_node(id),
            "LaunchSession context node {id} is not fully loaded in the parent context"
        );
    }
    Ok(())
}

fn authoritative_user_box_event(journal: &HistorySession, event: &Event) -> Option<EventId> {
    let EventKind::BoxCreated {
        box_id,
        owner: BoxOwner::User,
        ..
    } = &event.kind
    else {
        return None;
    };
    if *box_id != BoxId(event.id.0) {
        return None;
    }
    journal
        .state()
        .box_state(*box_id)
        .filter(|state| matches!(state.owner, BoxOwner::User))
        .map(|_| event.id)
}

fn unique_user_box_event(
    journal: &HistorySession,
    events: &[Event],
) -> anyhow::Result<Option<EventId>> {
    let ids = events
        .iter()
        .filter_map(|event| authoritative_user_box_event(journal, event))
        .collect::<Vec<_>>();
    anyhow::ensure!(
        ids.len() <= 1,
        "multiple authoritative user inputs appeared in one launch-authority interval"
    );
    Ok(ids.into_iter().next())
}

fn validate_user_turn_id(journal: &HistorySession, id: EventId) -> anyhow::Result<()> {
    let event = journal
        .state()
        .event(id)
        .context("restored launch user-turn event does not exist")?;
    anyhow::ensure!(
        authoritative_user_box_event(journal, event) == Some(id),
        "restored launch user-turn event is not an authoritative user BoxCreated event"
    );
    Ok(())
}

fn consume_launch_bootstrap_marker(
    orchestration: &mut Value,
    launch_bootstrap_pending: &mut bool,
) -> bool {
    if !*launch_bootstrap_pending {
        return false;
    }
    *launch_bootstrap_pending = false;
    if !orchestration.is_object() {
        *orchestration = json!({});
    }
    orchestration["launchBootstrapPending"] = json!(false);
    true
}

fn user_turn_launch_authority(
    user_turn: Option<EventId>,
    orchestration: &mut Value,
    launch_bootstrap_pending: &mut bool,
) -> Option<EventId> {
    if consume_launch_bootstrap_marker(orchestration, launch_bootstrap_pending) {
        None
    } else {
        user_turn
    }
}

fn reconcile_recovered_launch_authority(
    pending_turn: bool,
    recovered_user_turn: Option<EventId>,
    launch_provenance: &Value,
    orchestration: &mut Value,
    launch_bootstrap_pending: &mut bool,
    launch_user_turn_id: &mut Option<EventId>,
) {
    if let Some(recovered) = recovered_user_turn {
        if *launch_bootstrap_pending || !launch_provenance.is_null() {
            consume_launch_bootstrap_marker(orchestration, launch_bootstrap_pending);
            *launch_user_turn_id = None;
        } else {
            *launch_user_turn_id = Some(recovered);
        }
    }
    if !pending_turn || *launch_bootstrap_pending {
        *launch_user_turn_id = None;
    }
}

fn completed_invocation_ids(journal: &HistorySession) -> BTreeSet<String> {
    journal
        .state()
        .events
        .iter()
        .filter_map(|event| {
            let EventKind::ToolCompleted {
                invocation_id: Some(id),
                ..
            } = &event.kind
            else {
                return None;
            };
            Some(id.clone())
        })
        .collect()
}

fn pruned_launch_intents(
    intents: &[LaunchIntent],
    current_turn: Option<EventId>,
    completed: &BTreeSet<String>,
) -> Vec<LaunchIntent> {
    intents
        .iter()
        .filter(|intent| {
            Some(intent.user_turn_id) == current_turn || !completed.contains(&intent.invocation_id)
        })
        .cloned()
        .collect()
}

fn invocation_arguments<'a>(
    journal: &'a HistorySession,
    invocation_id: &str,
) -> anyhow::Result<&'a Value> {
    journal
        .state()
        .events
        .iter()
        .find_map(|event| {
            let EventKind::ToolInvoked {
                tool_name,
                arguments,
                invocation_id: Some(id),
                ..
            } = &event.kind
            else {
                return None;
            };
            (tool_name == LAUNCH_SESSION_TOOL && id == invocation_id).then_some(arguments)
        })
        .with_context(|| {
            format!(
                "launch intent {} has no matching ToolInvoked event",
                invocation_id
            )
        })
}

fn launch_success_json(session_id: &str, command_id: &str) -> anyhow::Result<String> {
    serde_json::to_string(&LaunchSuccess {
        session_id,
        command_id,
    })
    .context("serializing LaunchSession result")
}

fn tool_invocation_content(name: &str, arguments: &Value) -> anyhow::Result<BoxContent> {
    if name == LAUNCH_SESSION_TOOL {
        return Ok(BoxContent::text("LaunchSession"));
    }
    invocation_box_content(name, arguments)
}

pub struct Session {
    api: Service,
    subagent_codex_prompt: String,
    runtime: RuntimeModel,
    journal: HistorySession,
    plan: KwebPlan,
    pub session_type: String,
    pub channel: Value,
    pub free_time: Value,
    pub orchestration: Value,
    pub provenance_id: Option<String>,
    pub rust_lib_session_id: String,
    pub root_node_ids: Vec<String>,
    pub reference_root_node_ids: Vec<String>,
    pub started_at: String,
    pub transcript: Vec<Value>,
    pub pending_turn: bool,
    pub pending_external_event_id: Option<String>,
    pub completed: bool,
    pub rounds_used: u64,
    commit_receipt: Option<CommitReceipt>,
    commit_author: String,
    mode: AgentMode,
    source_session_type: Option<String>,
    group_context: Value,
    context: KwebContext,
    free_time_end_reason: Option<String>,
    fatal_persistence_error: Option<String>,
    active_provider_deadline: Option<DateTime<Utc>>,
    active_turn_deadline: Option<TurnDeadline>,
    provider_affinity: Option<ProviderAffinityState>,
    next_thread_reset_reason: Option<String>,
    ingress_deadline: Option<Instant>,
    previous_ingress_attempt_timed_out: bool,
    launch_provenance: Value,
    launch_context_node_ids: Vec<String>,
    launch_user_turn_id: Option<EventId>,
    launch_intents: Vec<LaunchIntent>,
    launch_bootstrap_pending: bool,
    turn_lease_slot: Option<Weak<TurnLeaseToken>>,
}

struct TurnLeaseToken;

struct TurnLease {
    token: Arc<TurnLeaseToken>,
}

impl TurnLease {
    fn acquire(slot: &mut Option<Weak<TurnLeaseToken>>) -> anyhow::Result<Self> {
        anyhow::ensure!(
            slot.as_ref().and_then(Weak::upgrade).is_none(),
            "a stepped session turn is already active"
        );
        let token = Arc::new(TurnLeaseToken);
        *slot = Some(Arc::downgrade(&token));
        Ok(Self { token })
    }

    fn validate(&self, slot: &Option<Weak<TurnLeaseToken>>) -> anyhow::Result<()> {
        anyhow::ensure!(
            slot.as_ref()
                .and_then(Weak::upgrade)
                .is_some_and(|active| Arc::ptr_eq(&self.token, &active)),
            "the stepped session turn is stale"
        );
        Ok(())
    }
}

struct PrimaryTurnState {
    accounting: Option<kcode_intelligence_chatend::TopLevelCall>,
    pending_freeform_write: Option<PendingFreeformWrite>,
    deadline_after_response: bool,
    operation_id: Uuid,
    prepared_cache: Option<PreparedCacheObservation>,
    provider_synchronized_after: Option<EventId>,
    restart_fresh_reason: Option<String>,
    exact_tool_result: bool,
    used_tool: bool,
    finish_requested: bool,
    emitted_response: bool,
    pending_capture: Option<Value>,
}

/// Owned serial state for one stepped Kennedy turn.
#[must_use]
pub struct SessionTurn {
    lease: TurnLease,
    user_id: String,
    completed_rounds: u64,
    round_limit: u64,
    state: PrimaryTurnState,
}

enum PendingInferenceAction {
    Start {
        runtime: kcode_agent_runtime::AgentRuntime,
        request: kcode_agent_runtime::SessionInferenceRequest,
    },
    Next {
        inference: kcode_agent_runtime::SessionInference,
    },
    Respond {
        inference: kcode_agent_runtime::SessionInference,
        call_id: String,
        result: kcode_codex_runtime_v2::ToolResult,
        stop: bool,
    },
}

/// An owned provider operation that can wait without borrowing a Session.
#[must_use]
pub struct PendingSessionInference {
    turn: SessionTurn,
    action: Box<PendingInferenceAction>,
}

enum SessionInferenceWakeKind {
    Event {
        inference: kcode_agent_runtime::SessionInference,
        event: anyhow::Result<Option<kcode_agent_runtime::SessionInferenceEvent>>,
    },
    StartFailed(anyhow::Error),
    RespondFailed {
        inference: kcode_agent_runtime::SessionInference,
        error: anyhow::Error,
    },
    RespondedStop {
        inference: kcode_agent_runtime::SessionInference,
    },
}

/// An owned provider result waiting to be serially applied to a Session.
#[must_use]
pub struct SessionInferenceWake {
    turn: SessionTurn,
    kind: SessionInferenceWakeKind,
}

/// One explicit boundary in a stepped primary Kennedy turn.
#[must_use]
pub enum TurnBoundary {
    /// Wait for provider work without borrowing the Session.
    Await(PendingSessionInference),
    /// A provider transition is applied and no later inference is prepared.
    Yield(SessionTurn),
    /// The logical turn has completed.
    Complete(Option<String>),
}

impl PendingSessionInference {
    /// Waits for one provider start, stream, or native response operation.
    pub async fn wait(self) -> SessionInferenceWake {
        let Self { turn, action } = self;
        let kind = match *action {
            PendingInferenceAction::Start { runtime, request } => {
                match runtime.start_session_inference(request).await {
                    Ok(mut inference) => {
                        let event = inference.next_event().await;
                        SessionInferenceWakeKind::Event { inference, event }
                    }
                    Err(error) => SessionInferenceWakeKind::StartFailed(error),
                }
            }
            PendingInferenceAction::Next { mut inference } => {
                let event = inference.next_event().await;
                SessionInferenceWakeKind::Event { inference, event }
            }
            PendingInferenceAction::Respond {
                mut inference,
                call_id,
                result,
                stop,
            } => match inference.respond(&call_id, result).await {
                Ok(()) if stop => SessionInferenceWakeKind::RespondedStop { inference },
                Ok(()) => {
                    let event = inference.next_event().await;
                    SessionInferenceWakeKind::Event { inference, event }
                }
                Err(error) => SessionInferenceWakeKind::RespondFailed { inference, error },
            },
        };
        SessionInferenceWake { turn, kind }
    }
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct ProviderAffinityState {
    continuation: kcode_intelligence_router::AgentContinuation,
    synchronized_event_id: EventId,
    material_fingerprint: String,
}

#[derive(Debug, Eq, PartialEq)]
enum NativeProviderResumePreparation {
    Continue { marker_lines: Vec<String> },
    RestartFresh { reason: String },
}

fn apply_prepared_provider_resume(
    provider_affinity: &mut Option<ProviderAffinityState>,
    next_thread_reset_reason: &mut Option<String>,
    prepared: PreparedProviderResume,
) -> NativeProviderResumePreparation {
    match prepared.thread_reset_reason {
        Some(reason) => {
            *provider_affinity = None;
            *next_thread_reset_reason = Some(reason.clone());
            NativeProviderResumePreparation::RestartFresh { reason }
        }
        None => NativeProviderResumePreparation::Continue {
            marker_lines: prepared.marker_lines,
        },
    }
}

fn restore_provider_affinity(
    restored: Option<&Value>,
    fresh_ingress_attempt: bool,
) -> anyhow::Result<Option<ProviderAffinityState>> {
    let state_version = restored
        .and_then(|state| state.get("stateVersion"))
        .and_then(Value::as_u64);
    if fresh_ingress_attempt || state_version != Some(CHECKPOINT_STATE_VERSION) {
        return Ok(None);
    }
    restored
        .and_then(|state| state.get("providerAffinity"))
        .filter(|value| !value.is_null())
        .cloned()
        .map(serde_json::from_value)
        .transpose()
        .context("restored provider affinity is invalid")
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum InputStage {
    Accepted,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ContextRecovery {
    NotNeeded,
    Recovered,
    Irreducible,
}

fn render_load_nodes_result(
    journal: &HistorySession,
    changed_box_ids: &[BoxId],
    footer_lines: &[String],
) -> anyhow::Result<String> {
    let changed_box_ids = changed_box_ids
        .iter()
        .map(ToString::to_string)
        .collect::<Vec<_>>();
    let projected_boxes = journal
        .state()
        .projection_with_footer_lines(footer_lines)
        .items
        .into_iter()
        .filter(|item| !item.marker)
        .map(|item| (item.box_id.to_string(), item.text))
        .collect::<Vec<_>>();
    render(RenderRequest::LoadNodes {
        changed_box_ids: &changed_box_ids,
        projected_boxes: &projected_boxes,
    })
}

fn provider_tool_result_with_context_footer(footer: &str, result: &str) -> String {
    render(RenderRequest::ProviderFooter { result, footer })
        .expect("provider-footer rendering is infallible")
}

fn completes_before_provider_resume(outcome: &kcode_agent_runtime::SessionToolOutcome) -> bool {
    outcome.stop || (outcome.ok && outcome.finish_after_round)
}

fn append_slow_tool_duration(text: &mut String, elapsed: Duration) {
    *text = render(RenderRequest::SlowTool { text, elapsed })
        .expect("slow-tool rendering is infallible");
}

fn log_primary_thread_observation(
    operation_id: Uuid,
    round: u64,
    requested_model: &str,
    prepared: &PreparedCacheObservation,
    provider_thread_id: Option<&str>,
    input_tokens: u64,
    cached_input_tokens: u64,
) {
    tracing::info!(
        affinity_scope = "primary",
        %operation_id,
        round,
        provider = prepared.provider,
        requested_model,
        model = prepared.model,
        thread_action = prepared.thread_action,
        provider_thread_id = provider_thread_id.unwrap_or(""),
        thread_reset_reason = prepared.thread_reset_reason.as_deref().unwrap_or(""),
        projection_hash = prepared.projection_hash,
        provider_input_hash = prepared.provider_input_hash,
        provider_input_bytes = prepared.provider_input_bytes,
        input_tokens,
        cached_input_tokens,
        "Provider thread-affinity observation"
    );
}

fn render_web_search_result(
    result: &kcode_intelligence_router::SearchResponse,
) -> anyhow::Result<String> {
    let sources = result
        .sources
        .iter()
        .map(|source| (source.title.clone(), source.url.clone()))
        .collect::<Vec<_>>();
    render(RenderRequest::WebSearch {
        answer: &result.answer,
        sources: &sources,
    })
}

fn render_web_fetch_result(
    result: &kcode_intelligence_router::FetchResponse,
) -> anyhow::Result<String> {
    render(RenderRequest::WebFetch {
        url: &result.url,
        title: result.title.as_deref(),
        content_type: &result.content_type,
        truncated: result.truncated,
        content: &result.content,
    })
}

fn render_media_annotation_result(
    object_id: &str,
    file_name: &str,
    content_type: &str,
    result: &kcode_intelligence_router::AnnotationResponse,
) -> anyhow::Result<String> {
    render(RenderRequest::MediaAnnotation {
        object_id,
        file_name,
        content_type,
        model: &result.model,
        complete: result.complete,
        incomplete_reason: result.incomplete_reason.as_deref(),
        text: &result.text,
    })
}

fn render_audio_transcription_result(
    object_id: &str,
    file_name: &str,
    content_type: &str,
    result: &kcode_intelligence_router::TranscriptionResponse,
) -> anyhow::Result<String> {
    render(RenderRequest::AudioTranscription {
        object_id,
        file_name,
        content_type,
        model: &result.model,
        text: &result.text,
    })
}

fn render_document_extraction_result(
    object_id: &str,
    file_name: &str,
    result: &kcode_intelligence_router::DocumentExtraction,
) -> anyhow::Result<String> {
    render(RenderRequest::DocumentExtraction {
        object_id,
        file_name,
        format: &result.format,
        characters: result.characters,
        truncated: result.truncated,
        text: &result.text,
    })
}

struct ToolCall {
    name: String,
    arguments: Value,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct TaskId {
    task_id: String,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct CategoryId {
    category_id: String,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct CategoryCall {
    category_id: String,
    #[serde(default)]
    offset: u64,
    #[serde(default = "task_page_limit")]
    limit: u32,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct EmptyCall {}

fn task_page_limit() -> u32 {
    50
}

struct RecordedToolInvocation {
    invocation_id: String,
    tool_instance: String,
    tool_name: String,
}

fn record_tool_completion_event(
    journal: &mut HistorySession,
    invocation: Option<&RecordedToolInvocation>,
    outcome: Value,
) -> anyhow::Result<EventId> {
    let (tool_instance, tool_name, invocation_id) = invocation
        .map(|invocation| {
            (
                invocation.tool_instance.clone(),
                invocation.tool_name.clone(),
                Some(invocation.invocation_id.clone()),
            )
        })
        .unwrap_or_else(|| ("call_ktool".into(), "call_ktool".into(), None));
    journal.record(
        now(),
        EventKind::ToolCompleted {
            tool_instance,
            tool_name,
            outcome,
            invocation_id,
        },
    )
}

fn ensure_tool_result_box(
    journal: &mut HistorySession,
    invocation: Option<&RecordedToolInvocation>,
    text: &str,
    ok: bool,
) -> anyhow::Result<String> {
    let Some(invocation) = invocation else {
        journal.create_box(
            now(),
            "Kennedy tool result",
            BoxOwner::Controller,
            BoxContent::text(text),
        )?;
        return Ok(text.to_owned());
    };
    let matches = journal
        .state()
        .boxes
        .values()
        .filter(|state| {
            matches!(state.owner, BoxOwner::Controller)
                && state
                    .canonical
                    .content
                    .metadata
                    .get("toolInvocationId")
                    .and_then(Value::as_str)
                    == Some(invocation.invocation_id.as_str())
        })
        .map(|state| {
            (
                state.id,
                state.canonical.content.text.clone(),
                state
                    .canonical
                    .content
                    .metadata
                    .get("toolResultOk")
                    .and_then(Value::as_bool),
            )
        })
        .collect::<Vec<_>>();
    anyhow::ensure!(
        matches.len() <= 1,
        "tool invocation {} has duplicate durable result boxes",
        invocation.invocation_id
    );
    if let Some((_box_id, stored_text, stored_ok)) = matches.into_iter().next() {
        anyhow::ensure!(
            stored_ok == Some(ok),
            "tool invocation {} has a result box with a conflicting outcome",
            invocation.invocation_id
        );
        return Ok(stored_text);
    }

    let mut content = BoxContent::text(text);
    content.metadata = json!({
        "toolInvocationId":invocation.invocation_id,
        "toolInstance":invocation.tool_instance,
        "toolName":invocation.tool_name,
        "toolResultOk":ok,
    });
    journal.create_box(now(), "Kennedy tool result", BoxOwner::Controller, content)?;
    Ok(text.to_owned())
}

fn complete_launch_reconciliation(
    journal: &mut HistorySession,
    invocation: &RecordedToolInvocation,
    result: Result<kcode_session_history::SessionLaunch, kcode_session_history::Error>,
) -> anyhow::Result<()> {
    if completed_invocation_ids(journal).contains(&invocation.invocation_id) {
        return Ok(());
    }
    let (ok, text) = match result {
        Ok(launch) => (
            true,
            launch_success_json(&launch.session_id, &launch.command_id)?,
        ),
        Err(error)
            if matches!(
                error.kind,
                HistoryErrorKind::InvalidInput | HistoryErrorKind::Conflict
            ) =>
        {
            (false, format!("LaunchSession failed: {}", error.message))
        }
        Err(error) => {
            anyhow::bail!(
                "LaunchSession reconciliation remains unresolved ({}): {}",
                error.kind.code(),
                error.message
            );
        }
    };
    let text = ensure_tool_result_box(journal, Some(invocation), &text, ok)?;
    record_tool_completion_event(journal, Some(invocation), json!({"ok":ok,"result":text}))?;
    Ok(())
}

struct PendingFreeformWrite {
    request: FreeformWrite,
    call_box_id: BoxId,
}

struct ToolOutcome {
    text: String,
    store_result: bool,
    ok: bool,
    end_session: bool,
    freeform_write: Option<FreeformWrite>,
    managed_source_snapshot: Option<SourceSnapshot>,
    exact_result: bool,
}

fn result_displays_snapshot(result: &str, snapshot: &SourceSnapshot) -> bool {
    result == snapshot.text
}

fn subagent_managed_write_fits(
    context: &SubagentContext,
    call: &ToolCall,
    budget: &kcode_agent_runtime::ContextBudget,
) -> bool {
    let Some(snapshot) = proposed_write_snapshot(&call.name, &call.arguments) else {
        return true;
    };
    let state = context.source_state(&snapshot);
    budget.fits_state(state.key, state.text)
}

struct KennedySubagentHost<'a> {
    session: &'a mut Session,
    context: SubagentContext,
    captures: HashMap<String, FreeformWrite>,
}

struct KennedySessionHost<'a, C> {
    session: &'a mut Session,
    checkpoint: &'a mut C,
    state: &'a mut PrimaryTurnState,
}

struct PreparedCacheObservation {
    cacheable_prefix_bytes: u64,
    expectation: CacheExpectation,
    material_fingerprint: String,
    projection_hash: String,
    logical_input: String,
    provider_input_hash: String,
    provider_input_bytes: u64,
    thread_action: String,
    thread_reset_reason: Option<String>,
    estimated_input_tokens: u64,
    raw_estimated_input_tokens: u64,
    provider: String,
    model: String,
}

fn is_kweb_mutation(name: &str) -> bool {
    matches!(
        name,
        "ConnectNodes" | "ConsolidateFanout" | "SetFixedConnection" | "CreateNode" | "UpdateNode"
    )
}

fn subagent_unavailable_reason(name: &str) -> Option<&'static str> {
    match name {
        LAUNCH_SESSION_TOOL => Some(
            "LaunchSession is unavailable inside a subagent. Only an authorized genuine parent user turn may launch a session.",
        ),
        "RunSubagent" => {
            Some("RunSubagent is unavailable inside a subagent. Only Kennedy may launch subagents.")
        }
        "EndSession" => Some(
            "EndSession is unavailable inside a subagent. A child cannot control the parent session lifecycle.",
        ),
        "DehydrateBoxes" | "SummarizeBox" | "HydrateBox" | "BoxesIntoObjects" => {
            Some("Parent box controls are unavailable inside a box-free subagent context.")
        }
        _ => None,
    }
}

fn ensure_plan_node_known(plan: &KwebPlan, context: &KwebContext, id: &str) -> anyhow::Result<()> {
    if id.starts_with("pending:") {
        anyhow::ensure!(
            plan.contains_pending(id),
            "pending node {id} is not part of this session"
        );
    } else {
        canonical_id(id)?;
        anyhow::ensure!(
            context.contains_full_node(id),
            "node {id} is not loaded; call LoadNodes first"
        );
    }
    Ok(())
}

fn kweb_mutation(
    decoded: DecodedTool,
    context: &KwebContext,
    plan: &KwebPlan,
    journal: &mut HistorySession,
) -> anyhow::Result<KwebMutation> {
    Ok(match decoded {
        DecodedTool::ConnectNodes(nodes) => KwebMutation::ConnectNodes(nodes),
        DecodedTool::ConsolidateFanout {
            parent,
            fanout,
            aggregator,
        } => KwebMutation::ConsolidateFanout {
            parent,
            fanout,
            aggregator,
        },
        DecodedTool::SetFixedConnection {
            parent,
            child,
            slot,
        } => KwebMutation::SetFixedConnection {
            parent,
            child,
            slot,
        },
        DecodedTool::CreateNode {
            parents,
            owner,
            short_name,
            short_description,
            long_description,
        } => {
            for id in parents.iter().chain(std::iter::once(&owner)) {
                if id != "self" && id != "unowned" {
                    ensure_plan_node_known(plan, context, id)?;
                }
            }
            KwebMutation::CreateNode {
                pending_id: journal.allocate_pending_node(now())?.to_string(),
                parents,
                owner,
                short_name,
                short_description,
                long_description,
            }
        }
        DecodedTool::UpdateNode {
            id,
            owner,
            short_name,
            short_description,
            long_description,
        } => KwebMutation::UpdateNode {
            id,
            owner,
            short_name,
            short_description,
            long_description,
        },
        _ => anyhow::bail!("decoded contract did not match a Kweb mutation"),
    })
}

fn connect_nodes_result_with_counts(
    result: String,
    plan: &KwebPlan,
    ids: &[String],
) -> anyhow::Result<String> {
    let (updates, creates) = plan.context_projection();
    let mut seen = BTreeSet::new();
    let mut counts = Vec::new();
    for id in ids {
        if !seen.insert(id.as_str()) {
            continue;
        }
        let count = updates
            .get(id)
            .map(|node| node.recent_connections.len())
            .or_else(|| {
                creates
                    .iter()
                    .find(|create| create.pending_id == id.as_str())
                    .map(|create| create.data.recent_connections.len())
            })
            .with_context(|| format!("ConnectNodes did not stage touched node {id}"))?;
        counts.push(format!("{id}: {count}"));
    }
    Ok(format!(
        "{result}\nPost-call recent connection counts: {}.",
        counts.join(", ")
    ))
}

fn execute_kweb_mutation(
    name: &str,
    decoded: DecodedTool,
    context: &KwebContext,
    plan: &mut KwebPlan,
    journal: &mut HistorySession,
) -> anyhow::Result<(String, Vec<String>)> {
    let mutation = kweb_mutation(decoded, context, plan, journal)
        .with_context(|| format!("decoded contract for {name} did not match its Kweb mutation"))?;
    let connect_nodes = match &mutation {
        KwebMutation::ConnectNodes(ids) => Some(ids.clone()),
        _ => None,
    };
    let referenced = referenced_pending_nodes(&mutation);
    let mut result = plan.apply(context, mutation)?;
    if let Some(ids) = connect_nodes {
        result = connect_nodes_result_with_counts(result, plan, &ids)?;
    }
    Ok((result, referenced))
}

fn inference_error_receipt(
    error: &anyhow::Error,
) -> Option<kcode_intelligence_router::UsageReceipt> {
    error.chain().find_map(|cause| {
        cause
            .downcast_ref::<kcode_intelligence_router::Error>()
            .and_then(|error| error.receipt().cloned())
    })
}

async fn record_unavailable_inference<C, F>(
    host: &mut KennedySessionHost<'_, C>,
    inference: &mut kcode_agent_runtime::SessionInference,
    round: u64,
) -> anyhow::Result<()>
where
    C: FnMut(Value) -> F + Send,
    F: Future<Output = anyhow::Result<()>> + Send,
{
    let receipt = inference.finish_unavailable()?;
    host.record(kcode_agent_runtime::SessionEvent::ProviderReceipt {
        round,
        usage: None,
        receipt,
        continuation: None,
    })
    .await
}

impl Session {
    /// Marks a fresh ingress attempt as following a prior timer expiry.
    pub fn mark_previous_ingress_attempt_timed_out(&mut self) {
        if matches!(self.mode, AgentMode::Ingress { .. })
            && !self.previous_ingress_attempt_timed_out
        {
            self.invalidate_active_stepped_turn();
            self.previous_ingress_attempt_timed_out = true;
        }
    }

    fn ingress_time_remaining(&mut self) -> anyhow::Result<Option<u64>> {
        if !matches!(self.mode, AgentMode::Ingress { .. }) {
            return Ok(None);
        }
        ingress_time_remaining_at(&mut self.ingress_deadline, Instant::now()).map(Some)
    }

    fn runtime_budget(&self) -> RuntimeBudget {
        let Some(provider_deadline) = self.active_provider_deadline else {
            return RuntimeBudget::default();
        };
        let mut time_limits = vec![TimeBudget {
            kind: TimeBudgetKind::ProviderCall,
            remaining: remaining_until(provider_deadline),
        }];
        if matches!(self.mode, AgentMode::FreeTime)
            && let Some(work_deadline) = deadline(&self.free_time)
        {
            time_limits.push(TimeBudget {
                kind: TimeBudgetKind::SelfTimeWork,
                remaining: remaining_until(work_deadline),
            });
        }
        if let Some(outer) = self.active_turn_deadline {
            time_limits.push(TimeBudget {
                kind: match outer.kind {
                    TurnDeadlineKind::Telegram => TimeBudgetKind::TelegramTurn,
                    TurnDeadlineKind::SelfTimeHardStop => TimeBudgetKind::SelfTimeHardStop,
                },
                remaining: remaining_until(outer.at),
            });
        }
        RuntimeBudget {
            rounds: Some(RoundBudget {
                used: self.rounds_used,
                limit: kcode_agent_runtime::DEFAULT_ROUND_LIMIT,
            }),
            time_limits,
        }
    }

    fn projection(&self) -> ContextProjection {
        self.journal
            .state()
            .projection_with_footer_lines(&self.runtime_budget().footer_lines())
    }

    fn provider_material_fingerprint(&self, tool_description: &str) -> String {
        let material = json!({
            "model":self.runtime.model,
            "reasoningEffort":self.runtime.reasoning_effort,
            "tool":"call_ktool",
            "toolDescription":tool_description,
        });
        hex::encode(Sha256::digest(
            serde_json::to_vec(&material).expect("provider material always serializes"),
        ))
    }

    fn begin_provider_call_budget(&mut self, timeout: Option<Duration>) {
        self.active_provider_deadline = timeout.and_then(|timeout| {
            chrono::Duration::from_std(timeout)
                .ok()
                .map(|timeout| Utc::now() + timeout)
        });
    }

    fn clear_turn_deadlines(&mut self) {
        self.active_provider_deadline = None;
        self.active_turn_deadline = None;
    }

    fn invalidate_active_stepped_turn(&mut self) {
        self.turn_lease_slot = None;
        self.clear_turn_deadlines();
    }

    fn synchronize_provider_known_events(&mut self) {
        if let Some(affinity) = self.provider_affinity.as_mut()
            && let Some(event) = self.journal.state().events.last()
        {
            affinity.synchronized_event_id = event.id;
        }
    }

    pub async fn new(
        api: Service,
        system_prompt: String,
        subagent_codex_prompt: String,
        runtime: RuntimeModel,
        started_at: String,
        mut options: SessionOptions,
        restored: Option<&Value>,
    ) -> anyhow::Result<Self> {
        if let Some(state) = restored {
            restore_session_type(&mut options, state);
            options.channel = state.get("channel").cloned().unwrap_or(options.channel);
            options.free_time = state.get("freeTime").cloned().unwrap_or(options.free_time);
            options.orchestration = state
                .get("orchestration")
                .cloned()
                .unwrap_or(options.orchestration);
        }
        if options.group_context.is_null() {
            options.group_context = options
                .channel
                .get("groupContext")
                .cloned()
                .unwrap_or(Value::Null);
        }
        options
            .reference_root_node_ids
            .retain(|id| !options.root_node_ids.contains(id));
        options.reference_root_node_ids.sort();
        options.reference_root_node_ids.dedup();

        DateTime::parse_from_rfc3339(&started_at).context("session start timestamp is invalid")?;
        if let Some(restored_started_at) = restored
            .and_then(|state| state.get("startedAt"))
            .and_then(Value::as_str)
        {
            anyhow::ensure!(
                restored_started_at == started_at,
                "restored session start timestamp changed"
            );
        }
        let rust_lib_session_id = restored
            .and_then(|state| state.get("rustLibSessionId"))
            .and_then(Value::as_str)
            .map(str::to_owned)
            .or(options.rust_lib_session_id.clone())
            .unwrap_or_else(|| format!("kennedy:{}", Uuid::new_v4()));
        let history_session_id = restored
            .and_then(|state| state.get("sessionId"))
            .and_then(Value::as_str)
            .map(str::to_owned);
        let source_session_type = options.source_session_type.clone().or_else(|| {
            restored
                .and_then(|state| state.get("sourceSessionType"))
                .and_then(Value::as_str)
                .map(str::to_owned)
        });
        let session_id = history_session_id
            .clone()
            .unwrap_or_else(|| Uuid::new_v4().to_string());
        let metadata = SessionMetadata {
            session_id: session_id.clone(),
            kind: session_kind(&options.session_type, &options.mode),
            created_at: started_at.clone(),
            effective_context_tokens: runtime.context_window_tokens,
            channel: options.channel.clone(),
        };
        let mut journal = if history_session_id.is_some() {
            api.history_session(metadata, &runtime.model)
                .with_context(|| {
                    format!(
                        "opening authoritative session {session_id} (legacy snapshots are intentionally unsupported)"
                    )
                })?
        } else {
            api.create_history_session(NewSession {
                kind: metadata.kind,
                created_at: metadata.created_at,
                effective_context_tokens: metadata.effective_context_tokens,
                channel: metadata.channel,
            })?
        };
        let checkpoint_event_count = restored
            .and_then(|state| state.get("eventCount"))
            .and_then(Value::as_u64)
            .map(|count| usize::try_from(count).context("checkpoint event count is too large"))
            .transpose()?
            .unwrap_or(journal.state().events.len());
        anyhow::ensure!(
            checkpoint_event_count <= journal.state().events.len(),
            "checkpoint event count is ahead of the durable journal"
        );
        let fresh_ingress_attempt = matches!(options.mode, AgentMode::Ingress { .. })
            && !journal.is_sealed()
            && journal.state().history_ingress_started;
        if fresh_ingress_attempt {
            journal.reset_history_ingress_attempt(now())?;
        }
        let launch_context_node_ids = restored
            .and_then(|state| state.get("launchContextNodeIds"))
            .cloned()
            .map(serde_json::from_value::<Vec<String>>)
            .transpose()
            .context("restored launch context node IDs are invalid")?
            .unwrap_or_default();
        validate_canonical_distinct_ids(&launch_context_node_ids, "context node")?;
        let mut context = KwebContext::with_fixed_connections(
            options.root_node_ids.clone(),
            api.loads_fixed_connections(),
        )
        .map_err(anyhow::Error::new)?;
        restore_kweb_context(&journal, &mut context)?;
        let plan = if fresh_ingress_attempt {
            KwebPlan::default()
        } else {
            KwebPlan::restore(
                restored.and_then(|state| state.get("kwebPlan")),
                journal_kweb_plan(&journal),
            )?
        };
        let transcript = kcode_kennedy_session_ingress::transcript_from_journal(&journal);
        let (pending_turn, pending_external_event_id) = restore_pending_turn(restored, &transcript);

        let needs_initialization = !journal
            .state()
            .boxes
            .values()
            .any(|state| matches!(state.owner, BoxOwner::System));
        let commit_receipt = restore_commit_receipt(restored)?;
        let commit_author = restored
            .and_then(|state| state.get("commitAuthor"))
            .and_then(Value::as_str)
            .map(str::to_owned)
            .unwrap_or_else(|| runtime.attribution());
        let provider_affinity = restore_provider_affinity(restored, fresh_ingress_attempt)?;
        let next_thread_reset_reason = (!fresh_ingress_attempt)
            .then(|| {
                restored
                    .and_then(|state| state.get("nextThreadResetReason"))
                    .and_then(Value::as_str)
                    .map(str::to_owned)
            })
            .flatten();
        if let Some(receipt) = &commit_receipt {
            journal.mark_completed(receipt.session_object_id.to_string());
        }
        let completed =
            journal.state().completed_session_object.is_some() || commit_receipt.is_some();
        let launch_provenance = restored
            .and_then(|state| state.get("launchProvenance"))
            .cloned()
            .unwrap_or(Value::Null);
        let mut launch_bootstrap_pending = restored
            .and_then(|state| state.get("orchestration"))
            .and_then(|value| value.get("launchBootstrapPending"))
            .and_then(Value::as_bool)
            .unwrap_or(!launch_provenance.is_null());
        let mut launch_user_turn_id = restored
            .and_then(|state| state.get("launchUserTurnId"))
            .filter(|value| !value.is_null())
            .cloned()
            .map(serde_json::from_value::<EventId>)
            .transpose()
            .context("restored launch user-turn ID is invalid")?;
        if let Some(id) = launch_user_turn_id {
            validate_user_turn_id(&journal, id)?;
        }
        let recovered_user_turn =
            if pending_turn && checkpoint_event_count < journal.state().events.len() {
                unique_user_box_event(&journal, &journal.state().events[checkpoint_event_count..])?
            } else {
                None
            };
        reconcile_recovered_launch_authority(
            pending_turn,
            recovered_user_turn,
            &launch_provenance,
            &mut options.orchestration,
            &mut launch_bootstrap_pending,
            &mut launch_user_turn_id,
        );
        let launch_intents = restored
            .and_then(|state| state.get("launchIntents"))
            .cloned()
            .map(serde_json::from_value::<Vec<LaunchIntent>>)
            .transpose()
            .context("restored launch intents are invalid")?
            .unwrap_or_default();
        let mut session = Self {
            api,
            subagent_codex_prompt,
            runtime,
            journal,
            plan,
            session_type: options.session_type,
            channel: options.channel,
            free_time: options.free_time,
            orchestration: options.orchestration,
            provenance_id: options.provenance_id,
            rust_lib_session_id,
            root_node_ids: options.root_node_ids,
            reference_root_node_ids: options.reference_root_node_ids,
            started_at,
            transcript,
            pending_turn,
            pending_external_event_id,
            completed,
            rounds_used: (!fresh_ingress_attempt)
                .then(|| {
                    restored
                        .and_then(|state| state.get("roundsUsed"))
                        .and_then(Value::as_u64)
                })
                .flatten()
                .unwrap_or_default(),
            commit_receipt,
            commit_author,
            mode: options.mode,
            source_session_type,
            group_context: options.group_context,
            context,
            free_time_end_reason: None,
            fatal_persistence_error: None,
            active_provider_deadline: None,
            active_turn_deadline: None,
            provider_affinity,
            next_thread_reset_reason,
            ingress_deadline: None,
            previous_ingress_attempt_timed_out: false,
            launch_provenance,
            launch_context_node_ids,
            launch_user_turn_id,
            launch_intents,
            launch_bootstrap_pending,
            turn_lease_slot: None,
        };
        session.validate_launch_intents()?;

        if session.journal.is_sealed() {
            session.provider_affinity = None;
            session.next_thread_reset_reason = None;
            anyhow::ensure!(
                !matches!(session.mode, AgentMode::Conversation),
                "a read-only conversation has an unexpectedly sealed session log"
            );
            if session.commit_receipt.is_none() {
                session.finalize_kweb_session()?;
            }
            session.completed = true;
            return Ok(session);
        }

        session.reconcile_launch_intents().await?;
        session.prune_launch_intents();
        session.repair_unfinished_tools()?;

        if needs_initialization {
            session.journal.create_box(
                now(),
                "Kennedy system prompt",
                BoxOwner::System,
                BoxContent::text(&system_prompt),
            )?;
            if session.session_type == "telegram-group" && !session.group_context.is_null() {
                session.journal.create_box(
                    now(),
                    "Telegram group context",
                    BoxOwner::Controller,
                    BoxContent::text(kcode_telegram_session_coordinator::format_group_context(
                        &session.group_context,
                    )),
                )?;
            }
            let identifiers = session.initial_context_identifiers();
            let invocation =
                session.record_tool_invocation("LoadNodes", json!({"identifiers":&identifiers}))?;
            let result =
                load_durable_batch(session.api.kmap(), &mut session.context, &identifiers)?;
            for id in &session.launch_context_node_ids {
                anyhow::ensure!(
                    session.context.contains_full_node(id),
                    "launched child context node {id} is unavailable"
                );
            }
            session.sync_kweb_boxes()?;
            session.record_tool_completion(
                Some(&invocation),
                json!({"ok":true,"automatic":true,"identifiers":identifiers,"result":result}),
            )?;
        } else {
            if !session.launch_context_node_ids.is_empty() {
                let identifiers = session.initial_context_identifiers();
                load_durable_batch(session.api.kmap(), &mut session.context, &identifiers)?;
                for id in &session.launch_context_node_ids {
                    anyhow::ensure!(
                        session.context.contains_full_node(id),
                        "launched child context node {id} is unavailable"
                    );
                }
            }
            session.sync_kweb_boxes()?;
        }
        if fresh_ingress_attempt {
            session.revalidate_loaded_nodes().await?;
            session.pending_turn = true;
        }
        if matches!(session.mode, AgentMode::Ingress { .. })
            && !session.completed
            && !session.journal.state().history_ingress_started
        {
            session.prepare_history_ingress(&system_prompt).await?;
        }
        Ok(session)
    }

    fn initial_context_identifiers(&self) -> Vec<String> {
        let mut identifiers = self.root_node_ids.clone();
        if !self.launch_context_node_ids.is_empty() {
            for id in self
                .reference_root_node_ids
                .iter()
                .chain(self.launch_context_node_ids.iter())
            {
                if !identifiers.contains(id) {
                    identifiers.push(id.clone());
                }
            }
        }
        identifiers
    }

    fn launch_session_authorized(&self) -> bool {
        self.pending_turn
            && self.launch_user_turn_id.is_some()
            && !self.launch_bootstrap_pending
            && matches!(self.mode, AgentMode::Conversation)
            && matches!(
                self.session_type.as_str(),
                "conversation" | "telegram" | "telegram-group"
            )
    }

    fn validate_launch_intents(&self) -> anyhow::Result<()> {
        let mut seen = BTreeSet::new();
        for intent in &self.launch_intents {
            Uuid::parse_str(&intent.invocation_id).with_context(|| {
                format!("launch intent {} has an invalid UUID", intent.invocation_id)
            })?;
            anyhow::ensure!(
                seen.insert(intent.invocation_id.as_str()),
                "duplicate launch intent {}",
                intent.invocation_id
            );
            validate_user_turn_id(&self.journal, intent.user_turn_id)?;
            DateTime::parse_from_rfc3339(&intent.started_at)
                .context("launch intent timestamp is invalid")?;
            anyhow::ensure!(
                intent.parent_session_id == self.journal.state().metadata.session_id,
                "launch intent parent session changed"
            );
            anyhow::ensure!(
                intent.effective_context_tokens > 0,
                "launch intent effective context size is invalid"
            );
            validate_canonical_distinct_ids(&intent.root_node_ids, "root node")?;
            validate_canonical_distinct_ids(
                &intent.reference_root_node_ids,
                "reference root node",
            )?;
            validate_canonical_distinct_ids(&intent.context_node_ids, "context node")?;
            let _ = invocation_arguments(&self.journal, &intent.invocation_id)?;
        }
        Ok(())
    }

    fn prune_launch_intents(&mut self) {
        let completed = completed_invocation_ids(&self.journal);
        self.launch_intents =
            pruned_launch_intents(&self.launch_intents, self.launch_user_turn_id, &completed);
    }

    fn unfinished_launch_intents(&self) -> Vec<&LaunchIntent> {
        let completed = completed_invocation_ids(&self.journal);
        self.launch_intents
            .iter()
            .filter(|intent| !completed.contains(&intent.invocation_id))
            .collect()
    }

    fn ensure_no_unfinished_launch_intents(&self) -> anyhow::Result<()> {
        let unfinished = self.unfinished_launch_intents();
        anyhow::ensure!(
            unfinished.is_empty(),
            "session has an unfinished intent-backed LaunchSession invocation"
        );
        Ok(())
    }

    fn repair_unfinished_tools(&mut self) -> anyhow::Result<()> {
        self.ensure_no_unfinished_launch_intents()?;
        self.journal.repair_unfinished_tools(now())?;
        Ok(())
    }

    fn prepare_launch_intent(
        &mut self,
        invocation: &RecordedToolInvocation,
        arguments: &LaunchSessionArguments,
    ) -> anyhow::Result<LaunchIntent> {
        anyhow::ensure!(
            self.launch_session_authorized(),
            "LaunchSession is unavailable without a genuine current user turn in an eligible conversation"
        );
        validate_loaded_launch_context(&self.context, &arguments.context_node_ids)?;
        if let Some(existing) = self
            .launch_intents
            .iter()
            .find(|intent| intent.invocation_id == invocation.invocation_id)
        {
            return Ok(existing.clone());
        }
        let user_turn_id = self
            .launch_user_turn_id
            .context("LaunchSession user-turn authority is missing")?;
        let current_count = self
            .launch_intents
            .iter()
            .filter(|intent| intent.user_turn_id == user_turn_id)
            .count();
        anyhow::ensure!(
            current_count < MAX_LAUNCH_INTENTS_PER_USER_TURN,
            "LaunchSession permits at most ten new sessions per genuine user turn"
        );
        let intent = LaunchIntent {
            invocation_id: invocation.invocation_id.clone(),
            user_turn_id,
            started_at: now(),
            parent_session_id: self.journal.state().metadata.session_id.clone(),
            effective_context_tokens: self.runtime.context_window_tokens,
            root_node_ids: self.root_node_ids.clone(),
            reference_root_node_ids: self.reference_root_node_ids.clone(),
            context_node_ids: arguments.context_node_ids.clone(),
        };
        self.launch_intents.push(intent.clone());
        Ok(intent)
    }

    fn launch_request(intent: &LaunchIntent, directive: &str) -> HistoryLaunchSession {
        let provenance = json!({
            "kind":"synthetic-launch-bootstrap",
            "denyLaunchSession":true,
            "parentSessionId":intent.parent_session_id,
            "parentInvocationId":intent.invocation_id,
            "parentUserTurnId":intent.user_turn_id,
        });
        HistoryLaunchSession {
            session_id: intent.invocation_id.clone(),
            started_at: intent.started_at.clone(),
            effective_context_tokens: intent.effective_context_tokens,
            channel: json!({"kind":"browser"}),
            state: json!({
                "sessionId":intent.invocation_id,
                "chatendMetadata":{
                    "sessionId":intent.invocation_id,
                    "kind":SessionKind::Conversation,
                    "createdAt":intent.started_at,
                    "effectiveContextTokens":intent.effective_context_tokens,
                    "channel":{"kind":"browser"},
                },
                "sessionType":"conversation",
                "channel":{"kind":"browser"},
                "freeTime":Value::Null,
                "orchestration":{
                    "owner":"backend",
                    "status":"idle",
                    "launchBootstrapPending":true,
                },
                "launchProvenance":provenance,
                "rootNodeIds":intent.root_node_ids,
                "referenceRootNodeIds":intent.reference_root_node_ids,
                "launchContextNodeIds":intent.context_node_ids,
                "startedAt":intent.started_at,
                "pendingTurn":false,
                "completed":false,
            }),
            initial_message: json!({
                "text":directive,
                "metadata":{
                    "launchProvenance":provenance,
                },
            }),
        }
    }

    async fn lower_launch(
        &self,
        intent: &LaunchIntent,
        arguments: &LaunchSessionArguments,
    ) -> Result<kcode_session_history::SessionLaunch, kcode_session_history::Error> {
        self.api
            .launch_session(Self::launch_request(intent, &arguments.directive))
            .await
    }

    async fn reconcile_launch_intents(&mut self) -> anyhow::Result<()> {
        let completed = completed_invocation_ids(&self.journal);
        let unfinished = self
            .launch_intents
            .iter()
            .filter(|intent| !completed.contains(&intent.invocation_id))
            .cloned()
            .collect::<Vec<_>>();
        for intent in unfinished {
            let raw = invocation_arguments(&self.journal, &intent.invocation_id)?.clone();
            let arguments = decode_launch_session_arguments(&raw)?;
            let invocation = RecordedToolInvocation {
                invocation_id: intent.invocation_id.clone(),
                tool_instance: tool_instance_for_invocation(
                    LAUNCH_SESSION_TOOL,
                    &intent.invocation_id,
                ),
                tool_name: LAUNCH_SESSION_TOOL.into(),
            };
            let result = self.lower_launch(&intent, &arguments).await;
            complete_launch_reconciliation(&mut self.journal, &invocation, result)?;
        }
        Ok(())
    }

    async fn prepare_history_ingress(&mut self, prompt: &str) -> anyhow::Result<()> {
        let cost_at_ingress = self.projection().status;
        if !self.journal.state().source_terminated {
            self.journal.record(
                now(),
                EventKind::SourceTerminated {
                    reason: "history_ingress".into(),
                },
            )?;
        }
        let system_box = self
            .journal
            .state()
            .boxes
            .values()
            .find(|state| matches!(state.owner, BoxOwner::System))
            .map(|state| state.id)
            .context("session has no system-prompt box")?;
        self.journal
            .update_box(now(), system_box, BoxContent::text(prompt))?;
        let ingress_kind = session_kind(&self.session_type, &self.mode);
        if self.journal.state().metadata.effective_context_tokens
            != self.runtime.context_window_tokens
            || self.journal.state().metadata.kind != ingress_kind
        {
            self.journal
                .configure_context(ingress_kind, self.runtime.context_window_tokens);
        }
        self.journal.create_box(
            now(),
            "Session cost at ingress",
            BoxOwner::Controller,
            BoxContent::text(cost_summary(
                "session cost before history ingress",
                cost_at_ingress.estimated_cost_usd_nanos,
                cost_at_ingress.unpriced_provider_calls,
            )),
        )?;
        self.revalidate_loaded_nodes().await?;
        match kcode_history_ingress_context::prepare(&mut self.journal, now())? {
            HistoryIngressContextOutcome::Ready => {}
            HistoryIngressContextOutcome::OverCapacity {
                estimated_tokens,
                target_tokens,
            } => {
                self.journal.record(
                    now(),
                    EventKind::Note {
                        label: INGRESS_FORCE_COMMIT_NOTE.into(),
                        value: json!({
                            "reason":"fully_dehydrated_context_above_initial_target",
                            "estimatedTokens":estimated_tokens,
                            "initialTargetTokens":target_tokens,
                        }),
                    },
                )?;
                self.clear_launch_turn_authority();
                self.pending_turn = false;
                self.finalize_kweb_session()?;
                self.completed = true;
                return Ok(());
            }
        }
        self.journal
            .record(now(), EventKind::HistoryIngressStarted)?;
        self.pending_turn = true;
        Ok(())
    }

    async fn revalidate_loaded_nodes(&mut self) -> anyhow::Result<()> {
        let direct = self.context.loaded_node_ids().to_vec();
        load_durable_batch(self.api.kmap(), &mut self.context, &direct)?;
        self.sync_kweb_boxes()?;
        Ok(())
    }

    fn stage_user_input(&mut self, text: &str, metadata: &Value) -> Option<InputStage> {
        let recorded_at = now();
        let result = (|| -> anyhow::Result<Option<InputStage>> {
            let Some(staged) = kcode_kennedy_session_ingress::stage_user_input(
                &mut self.journal,
                text,
                metadata,
                &recorded_at,
            )?
            else {
                return Ok(None);
            };
            self.transcript.push(staged.transcript);
            self.recover_context_overflow(staged.external_event_id.as_deref(), &[])?;
            Ok(Some(InputStage::Accepted))
        })();
        match result {
            Ok(stage) => stage,
            Err(error) => {
                self.fatal_persistence_error = Some(error.to_string());
                tracing::error!(error=%error, "Could not durably stage session input");
                Some(InputStage::Accepted)
            }
        }
    }

    pub fn append_final_user_message(&mut self, text: &str, metadata: &Value) -> bool {
        let accepted = self.stage_user_input(text, metadata).is_some();
        if accepted {
            self.invalidate_active_stepped_turn();
        }
        accepted
    }

    pub fn stage_source_message(
        &mut self,
        kennedy: bool,
        text: &str,
        metadata: Value,
    ) -> anyhow::Result<()> {
        let staged = kcode_kennedy_session_ingress::stage_source_input(
            &mut self.journal,
            kennedy,
            text,
            metadata,
            &now(),
        )?;
        self.invalidate_active_stepped_turn();
        self.transcript.push(staged.transcript);
        self.recover_context_overflow(staged.external_event_id.as_deref(), &[])?;
        Ok(())
    }

    pub fn answer_for_external_event(&self, id: &str) -> Option<&Value> {
        self.transcript.iter().rev().find(|entry| {
            is_terminal_external_response(entry)
                && entry.get("externalEventId").and_then(Value::as_str) == Some(id)
        })
    }

    pub fn responses_for_external_event(&self, id: &str) -> Vec<&Value> {
        self.transcript
            .iter()
            .filter(|entry| {
                matches!(
                    entry.get("role").and_then(Value::as_str),
                    Some("kennedy" | "system")
                ) && entry.get("externalEventId").and_then(Value::as_str) == Some(id)
            })
            .collect()
    }

    pub fn resolve_object(&mut self, object_id: &str) -> anyhow::Result<ResolvedObject> {
        let api = self.api.clone();
        kcode_kennedy_session_objects::resolve_object(
            &mut self.journal,
            object_id,
            move |canonical_id| api.kmap_file(canonical_id).map_err(Into::into),
        )
    }

    fn resolve_media_object(&mut self, object_id: &str) -> anyhow::Result<ResolvedObject> {
        let api = self.api.clone();
        kcode_kennedy_session_objects::resolve_media_object(
            &mut self.journal,
            object_id,
            MAX_MEDIA_ENRICHMENT_BYTES,
            move |canonical_id| api.kmap_file(canonical_id).map_err(Into::into),
        )
    }

    fn resolve_image_object(
        &mut self,
        object_id: &str,
    ) -> anyhow::Result<(Vec<u8>, String, String)> {
        let resolved = self.resolve_media_object(object_id)?;
        anyhow::ensure!(
            resolved.media_type.starts_with("image/"),
            "GenerateImage reference {object_id} is not an image"
        );
        Ok((resolved.bytes, resolved.file_name, resolved.media_type))
    }

    fn recover_context_overflow(
        &mut self,
        external_event_id: Option<&str>,
        pinned_box_ids: &[BoxId],
    ) -> anyhow::Result<ContextRecovery> {
        let projection = self.projection();
        let target_tokens = self.journal.state().active_context_limit();
        if projection.estimated_tokens <= target_tokens {
            return Ok(ContextRecovery::NotNeeded);
        }
        let projection_hash = hex::encode(Sha256::digest(projection.render().as_bytes()));
        let already_irreducible = self
            .journal
            .state()
            .events
            .iter()
            .rev()
            .find_map(|event| match &event.kind {
                EventKind::Note { label, value } if label == "context_overflow_recovery" => {
                    Some(value)
                }
                _ => None,
            })
            .is_some_and(|value| {
                value.get("irreducible").and_then(Value::as_bool) == Some(true)
                    && value.get("limitTokens").and_then(Value::as_u64) == Some(target_tokens)
                    && value.get("projectionHash").and_then(Value::as_str)
                        == Some(projection_hash.as_str())
            });
        if already_irreducible {
            return Ok(ContextRecovery::Irreducible);
        }

        let before_tokens = projection.estimated_tokens;
        let mut metadata = json!({
            "transcriptRole":"system",
            "contextOverflowWarning":true,
            "projectedTokens":before_tokens,
            "limitTokens":target_tokens,
        });
        if let Some(id) = external_event_id {
            metadata["externalEventId"] = json!(id);
        }
        let warning_box_id = self.journal.create_box(
            now(),
            CONTEXT_OVERFLOW_WARNING_BOX_NAME,
            BoxOwner::Controller,
            BoxContent {
                text: CONTEXT_OVERFLOW_WARNING.into(),
                objects: Vec::new(),
                metadata,
            },
        )?;
        let mut transcript = json!({
            "role":"system",
            "content":CONTEXT_OVERFLOW_WARNING,
            "contextOverflowWarning":true,
        });
        if let Some(id) = external_event_id {
            transcript["externalEventId"] = json!(id);
        }
        self.transcript.push(transcript);

        let mut pins = pinned_box_ids.to_vec();
        if !pins.contains(&warning_box_id) {
            pins.push(warning_box_id);
        }
        let outcome = kcode_history_ingress_context::recover(&mut self.journal, now(), &pins)?;
        let (dehydrated_box_ids, estimated_tokens, target_tokens, irreducible) = match outcome {
            ContextRecoveryOutcome::Recovered {
                dehydrated_box_ids,
                estimated_tokens,
                target_tokens,
            } => (dehydrated_box_ids, estimated_tokens, target_tokens, false),
            ContextRecoveryOutcome::OverCapacity {
                dehydrated_box_ids,
                estimated_tokens,
                target_tokens,
            } => (dehydrated_box_ids, estimated_tokens, target_tokens, true),
        };
        let final_projection_hash =
            hex::encode(Sha256::digest(self.projection().render().as_bytes()));
        self.journal.record(
            now(),
            EventKind::Note {
                label: "context_overflow_recovery".into(),
                value: json!({
                    "beforeTokens":before_tokens,
                    "estimatedTokens":estimated_tokens,
                    "limitTokens":target_tokens,
                    "dehydratedBoxIds":dehydrated_box_ids,
                    "irreducible":irreducible,
                    "projectionHash":final_projection_hash,
                }),
            },
        )?;
        if irreducible {
            if matches!(self.mode, AgentMode::Ingress { .. }) {
                self.request_ingress_force_commit(
                    "irreducible_context_overflow",
                    estimated_tokens,
                )?;
            } else if !self.journal.state().source_terminated {
                self.journal.record(
                    now(),
                    EventKind::SourceTerminated {
                        reason: "irreducible_context_overflow".into(),
                    },
                )?;
            }
            Ok(ContextRecovery::Irreducible)
        } else {
            Ok(ContextRecovery::Recovered)
        }
    }

    fn request_ingress_force_commit(
        &mut self,
        reason: &str,
        projected_tokens: u64,
    ) -> anyhow::Result<()> {
        if self.ingress_force_commit_requested() {
            return Ok(());
        }
        self.journal.record(
            now(),
            EventKind::Note {
                label: INGRESS_FORCE_COMMIT_NOTE.into(),
                value: json!({
                    "reason":reason,
                    "projectedTokens":projected_tokens,
                    "limitTokens":self.journal.state().ingress_context_limit(),
                }),
            },
        )?;
        Ok(())
    }

    fn ingress_force_commit_requested(&self) -> bool {
        self.journal
            .state()
            .current_ingress_attempt_events()
            .iter()
            .rev()
            .any(|event| {
                matches!(
                    &event.kind,
                    EventKind::Note { label, .. } if label == INGRESS_FORCE_COMMIT_NOTE
                )
            })
    }

    pub fn requires_history_ingress(&self) -> bool {
        matches!(self.mode, AgentMode::Conversation) && self.journal.state().source_terminated
    }

    pub fn stage_free_time_opening(&mut self) -> bool {
        if self.pending_turn {
            return false;
        }
        self.invalidate_active_stepped_turn();
        self.launch_user_turn_id = None;
        self.prune_launch_intents();
        let mut blocks = vec![
            render(RenderRequest::FreeTimeOpening {
                free_time: &self.free_time,
            })
            .expect("free-time opening rendering is infallible"),
        ];
        if let Some(message) = self
            .free_time
            .get("handoffMessage")
            .and_then(Value::as_str)
            .filter(|message| !message.trim().is_empty())
        {
            blocks.push(format!(
                "Message from the previous self-time session:\n\n{message}"
            ));
        }
        let Some(stage) = self.stage_user_input(&blocks.join("\n\n"), &json!({"kind":"self-time"}))
        else {
            return false;
        };
        self.pending_turn = matches!(stage, InputStage::Accepted);
        true
    }

    pub fn stage_wakeup_opening(&mut self) -> anyhow::Result<bool> {
        if self.pending_turn {
            return Ok(false);
        }
        self.invalidate_active_stepped_turn();
        self.launch_user_turn_id = None;
        self.prune_launch_intents();
        let marker = self
            .channel
            .get("wakeupMarker")
            .and_then(Value::as_str)
            .context("wakeup session is missing its acquired time marker")?;
        let marker = DateTime::parse_from_rfc3339(marker)
            .context("wakeup session has an invalid acquired time marker")?
            .with_timezone(&Utc);
        let text = render(RenderRequest::WakeupOpening { marker })?;
        let Some(stage) = self.stage_user_input(
            &text,
            &json!({"kind":"wakeup","wakeupMarker":marker.to_rfc3339()}),
        ) else {
            return Ok(false);
        };
        self.pending_turn = matches!(stage, InputStage::Accepted);
        Ok(true)
    }

    pub fn begin_user_turn(&mut self, text: &str, metadata: &Value) -> bool {
        if self.pending_turn {
            return false;
        }
        self.invalidate_active_stepped_turn();
        self.launch_user_turn_id = None;
        self.prune_launch_intents();
        let first_event = self.journal.state().events.len();
        let Some(stage) = self.stage_user_input(text, metadata) else {
            return false;
        };
        debug_assert_eq!(stage, InputStage::Accepted);
        let user_turn =
            unique_user_box_event(&self.journal, &self.journal.state().events[first_event..])
                .ok()
                .flatten();
        self.launch_user_turn_id = user_turn_launch_authority(
            user_turn,
            &mut self.orchestration,
            &mut self.launch_bootstrap_pending,
        );
        self.rounds_used = 0;
        self.pending_turn = true;
        self.pending_external_event_id = metadata
            .get("externalEventId")
            .and_then(Value::as_str)
            .map(str::to_owned);
        true
    }

    fn clear_launch_turn_authority(&mut self) {
        self.launch_user_turn_id = None;
        self.prune_launch_intents();
    }

    pub fn reset_exhausted_turn_rounds_for_retry(&mut self) {
        if matches!(self.mode, AgentMode::Conversation)
            && self.rounds_used >= kcode_agent_runtime::DEFAULT_ROUND_LIMIT
        {
            self.invalidate_active_stepped_turn();
            self.rounds_used = 0;
        }
    }

    pub fn interrupt_current_turn(&mut self) -> anyhow::Result<()> {
        self.ensure_no_unfinished_launch_intents()?;
        self.invalidate_active_stepped_turn();
        self.provider_affinity = None;
        self.next_thread_reset_reason = Some("prior_provider_turn_interrupted".into());
        self.repair_unfinished_tools()?;
        let notice = "The user stopped this agent turn.";
        let mut metadata = json!({"transcriptRole":"system","userStopped":true});
        let mut transcript_entry = json!({
            "role":"system",
            "content":notice,
            "userStopped":true,
        });
        if let Some(external_event_id) = &self.pending_external_event_id {
            metadata["externalEventId"] = json!(external_event_id);
            transcript_entry["externalEventId"] = json!(external_event_id);
        }
        self.journal.create_box(
            now(),
            "Turn stopped",
            BoxOwner::Controller,
            BoxContent {
                text: notice.into(),
                objects: Vec::new(),
                metadata,
            },
        )?;
        self.transcript.push(transcript_entry);
        self.pending_turn = false;
        self.pending_external_event_id = None;
        self.clear_launch_turn_authority();
        self.orchestration =
            json!({"owner":"backend","status":"idle","lastOutcome":"user-stopped"});
        Ok(())
    }

    /// Acquires one pending logical turn for stepped execution.
    pub fn begin_pending_turn(
        &mut self,
        operation_id: Uuid,
        turn_deadline: Option<TurnDeadline>,
    ) -> anyhow::Result<Option<SessionTurn>> {
        if let Some(error) = self.fatal_persistence_error.take() {
            anyhow::bail!("session journal write failed: {error}");
        }
        if !self.pending_turn {
            return Ok(None);
        }
        let user_id = self
            .root_node_ids
            .first()
            .context("session has no user root for intelligence accounting")?
            .clone();
        let lease = TurnLease::acquire(&mut self.turn_lease_slot)?;
        self.active_provider_deadline = None;
        self.active_turn_deadline = turn_deadline;
        Ok(Some(SessionTurn {
            lease,
            user_id,
            completed_rounds: self.rounds_used,
            round_limit: kcode_agent_runtime::DEFAULT_ROUND_LIMIT,
            state: PrimaryTurnState {
                accounting: None,
                pending_freeform_write: None,
                deadline_after_response: false,
                operation_id,
                prepared_cache: None,
                provider_synchronized_after: None,
                restart_fresh_reason: None,
                exact_tool_result: false,
                used_tool: false,
                finish_requested: false,
                emitted_response: false,
                pending_capture: None,
            },
        }))
    }

    async fn finish_stepped_turn<C, F>(
        &mut self,
        turn: SessionTurn,
        result: Option<String>,
        checkpoint: &mut C,
    ) -> anyhow::Result<TurnBoundary>
    where
        C: FnMut(Value) -> F + Send,
        F: Future<Output = anyhow::Result<()>> + Send,
    {
        turn.lease.validate(&self.turn_lease_slot)?;
        self.turn_lease_slot = None;
        self.clear_turn_deadlines();
        let output = match self.mode {
            AgentMode::Conversation => {
                if self.journal.state().source_terminated {
                    self.provider_affinity = None;
                    self.next_thread_reset_reason = None;
                    self.pending_turn = false;
                    self.pending_external_event_id = None;
                    self.clear_launch_turn_authority();
                    checkpoint(self.snapshot()?).await?;
                    None
                } else {
                    let Some(answer) = result else {
                        if self
                            .pending_external_event_id
                            .as_deref()
                            .and_then(|id| self.answer_for_external_event(id))
                            .is_some()
                        {
                            self.pending_turn = false;
                            self.pending_external_event_id = None;
                            self.clear_launch_turn_authority();
                            checkpoint(self.snapshot()?).await?;
                            return Ok(TurnBoundary::Complete(None));
                        }
                        anyhow::bail!(
                            "Kennedy ended a conversational turn without an assistant response"
                        );
                    };
                    self.pending_turn = false;
                    self.pending_external_event_id = None;
                    self.clear_launch_turn_authority();
                    checkpoint(self.snapshot()?).await?;
                    Some(answer)
                }
            }
            AgentMode::FreeTime | AgentMode::Wakeup | AgentMode::Ingress { .. } => {
                self.pending_turn = false;
                self.pending_external_event_id = None;
                self.clear_launch_turn_authority();
                self.finalize_kweb_session()?;
                self.completed = true;
                checkpoint(self.snapshot()?).await?;
                None
            }
        };
        Ok(TurnBoundary::Complete(output))
    }

    /// Prepares the next provider inference from a stepped logical turn.
    pub async fn advance_pending_turn<C, F>(
        &mut self,
        turn: SessionTurn,
        checkpoint: &mut C,
    ) -> anyhow::Result<TurnBoundary>
    where
        C: FnMut(Value) -> F + Send,
        F: Future<Output = anyhow::Result<()>> + Send,
    {
        turn.lease.validate(&self.turn_lease_slot)?;
        let result = self.advance_pending_turn_validated(turn, checkpoint).await;
        if result.is_err() {
            self.clear_turn_deadlines();
        }
        result
    }

    async fn advance_pending_turn_validated<C, F>(
        &mut self,
        mut turn: SessionTurn,
        checkpoint: &mut C,
    ) -> anyhow::Result<TurnBoundary>
    where
        C: FnMut(Value) -> F + Send,
        F: Future<Output = anyhow::Result<()>> + Send,
    {
        if turn.completed_rounds >= turn.round_limit {
            let runtime = self.api.agent_runtime();
            let operation_id = turn.state.operation_id;
            let answer = {
                let mut host = KennedySessionHost {
                    session: self,
                    checkpoint,
                    state: &mut turn.state,
                };
                runtime
                    .run_session(
                        kcode_agent_runtime::SessionRunRequest {
                            user_id: turn.user_id.clone(),
                            operation_id,
                            rounds_used: turn.round_limit,
                            round_limit: turn.round_limit,
                        },
                        &mut host,
                    )
                    .await?
            };
            return self.finish_stepped_turn(turn, answer, checkpoint).await;
        }

        let round = turn.completed_rounds + 1;
        let operation_id = turn.state.operation_id;
        let runtime = self.api.agent_runtime();
        let prepared = {
            let mut host = KennedySessionHost {
                session: self,
                checkpoint,
                state: &mut turn.state,
            };
            match host.prepare_round(round).await? {
                kcode_agent_runtime::RoundPreparation::Run(prepared) => {
                    let manifest_hash = hex::encode(Sha256::digest(prepared.input.as_bytes()));
                    host.record(kcode_agent_runtime::SessionEvent::InferenceSubmitted {
                        round,
                        manifest_hash,
                        model: prepared.model.clone(),
                    })
                    .await?;
                    prepared
                }
                kcode_agent_runtime::RoundPreparation::Complete(answer) => {
                    return self.finish_stepped_turn(turn, answer, checkpoint).await;
                }
            }
        };
        turn.completed_rounds = round;
        Ok(TurnBoundary::Await(PendingSessionInference {
            turn,
            action: Box::new(PendingInferenceAction::Start {
                runtime,
                request: kcode_agent_runtime::SessionInferenceRequest {
                    user_id: self
                        .root_node_ids
                        .first()
                        .context("session has no user root for intelligence accounting")?
                        .clone(),
                    operation_id,
                    round,
                    prepared,
                },
            }),
        }))
    }

    /// Applies one owned provider wake to the serial Session state.
    pub async fn apply_inference_wake<C, F>(
        &mut self,
        wake: SessionInferenceWake,
        checkpoint: &mut C,
    ) -> anyhow::Result<TurnBoundary>
    where
        C: FnMut(Value) -> F + Send,
        F: Future<Output = anyhow::Result<()>> + Send,
    {
        wake.turn.lease.validate(&self.turn_lease_slot)?;
        let result = self.apply_inference_wake_validated(wake, checkpoint).await;
        if result.is_err() {
            self.clear_turn_deadlines();
        }
        result
    }

    async fn apply_inference_wake_validated<C, F>(
        &mut self,
        wake: SessionInferenceWake,
        checkpoint: &mut C,
    ) -> anyhow::Result<TurnBoundary>
    where
        C: FnMut(Value) -> F + Send,
        F: Future<Output = anyhow::Result<()>> + Send,
    {
        let SessionInferenceWake { mut turn, kind } = wake;
        let round = turn.completed_rounds;
        let operation_id = turn.state.operation_id;
        match kind {
            SessionInferenceWakeKind::StartFailed(error) => {
                let mut host = KennedySessionHost {
                    session: self,
                    checkpoint,
                    state: &mut turn.state,
                };
                if let Some(receipt) = inference_error_receipt(&error) {
                    host.record(kcode_agent_runtime::SessionEvent::ProviderReceipt {
                        round,
                        usage: None,
                        receipt: Box::new(receipt),
                        continuation: None,
                    })
                    .await?;
                }
                Err(error)
            }
            SessionInferenceWakeKind::RespondFailed {
                mut inference,
                error,
            } => {
                let mut host = KennedySessionHost {
                    session: self,
                    checkpoint,
                    state: &mut turn.state,
                };
                record_unavailable_inference(&mut host, &mut inference, round).await?;
                Err(error)
            }
            SessionInferenceWakeKind::RespondedStop { inference } => {
                drop(inference);
                self.finish_stepped_turn(turn, None, checkpoint).await
            }
            SessionInferenceWakeKind::Event {
                mut inference,
                event,
            } => {
                let event = match event {
                    Ok(Some(event)) => event,
                    Ok(None) => {
                        let mut host = KennedySessionHost {
                            session: self,
                            checkpoint,
                            state: &mut turn.state,
                        };
                        record_unavailable_inference(&mut host, &mut inference, round).await?;
                        anyhow::bail!("provider ended without a terminal turn event");
                    }
                    Err(error) => {
                        let mut host = KennedySessionHost {
                            session: self,
                            checkpoint,
                            state: &mut turn.state,
                        };
                        if let Some(receipt) = inference_error_receipt(&error) {
                            host.record(kcode_agent_runtime::SessionEvent::ProviderReceipt {
                                round,
                                usage: None,
                                receipt: Box::new(receipt),
                                continuation: None,
                            })
                            .await?;
                        }
                        return Err(error);
                    }
                };

                match event {
                    kcode_agent_runtime::SessionInferenceEvent::ProviderInput { context } => {
                        let mut host = KennedySessionHost {
                            session: self,
                            checkpoint,
                            state: &mut turn.state,
                        };
                        host.record(kcode_agent_runtime::SessionEvent::ProviderInput {
                            round,
                            context,
                        })
                        .await?;
                        Ok(TurnBoundary::Await(PendingSessionInference {
                            turn,
                            action: Box::new(PendingInferenceAction::Next { inference }),
                        }))
                    }
                    kcode_agent_runtime::SessionInferenceEvent::UsageUpdated { usage } => {
                        let mut host = KennedySessionHost {
                            session: self,
                            checkpoint,
                            state: &mut turn.state,
                        };
                        host.record(kcode_agent_runtime::SessionEvent::UsageUpdated {
                            round,
                            usage,
                        })
                        .await?;
                        Ok(TurnBoundary::Await(PendingSessionInference {
                            turn,
                            action: Box::new(PendingInferenceAction::Next { inference }),
                        }))
                    }
                    kcode_agent_runtime::SessionInferenceEvent::ToolCall { call_id, call } => {
                        turn.state.used_tool = true;
                        let call =
                            call.map_err(|error| anyhow::anyhow!("Invalid Ktool call: {error}"));
                        let mut host = KennedySessionHost {
                            session: self,
                            checkpoint,
                            state: &mut turn.state,
                        };
                        let outcome = match host.execute_tool(call, operation_id).await {
                            Ok(outcome) => outcome,
                            Err(error) => {
                                record_unavailable_inference(&mut host, &mut inference, round)
                                    .await?;
                                return Err(error);
                            }
                        };
                        let resume = match host.prepare_provider_resume(outcome).await {
                            Ok(resume) => resume,
                            Err(error) => {
                                record_unavailable_inference(&mut host, &mut inference, round)
                                    .await?;
                                return Err(error);
                            }
                        };
                        match resume {
                            kcode_agent_runtime::ProviderResume::Continue(mut outcome) => {
                                host.state.finish_requested |=
                                    outcome.ok && outcome.finish_after_round;
                                host.state.emitted_response |=
                                    outcome.ok && outcome.emitted_response;
                                host.state.pending_capture = outcome.capture.take();
                                let stop = outcome.stop;
                                let result = if outcome.ok {
                                    kcode_codex_runtime_v2::ToolResult::success(outcome.text)
                                } else {
                                    kcode_codex_runtime_v2::ToolResult::failure(outcome.text)
                                };
                                Ok(TurnBoundary::Await(PendingSessionInference {
                                    turn,
                                    action: Box::new(PendingInferenceAction::Respond {
                                        inference,
                                        call_id,
                                        result,
                                        stop,
                                    }),
                                }))
                            }
                            kcode_agent_runtime::ProviderResume::Complete(answer) => {
                                record_unavailable_inference(&mut host, &mut inference, round)
                                    .await?;
                                self.finish_stepped_turn(turn, answer, checkpoint).await
                            }
                            kcode_agent_runtime::ProviderResume::RestartFresh => {
                                record_unavailable_inference(&mut host, &mut inference, round)
                                    .await?;
                                Ok(TurnBoundary::Yield(turn))
                            }
                        }
                    }
                    kcode_agent_runtime::SessionInferenceEvent::Completed {
                        answer,
                        usage,
                        receipt,
                        continuation,
                    } => {
                        drop(inference);
                        let mut host = KennedySessionHost {
                            session: self,
                            checkpoint,
                            state: &mut turn.state,
                        };
                        host.record(kcode_agent_runtime::SessionEvent::ProviderReceipt {
                            round,
                            usage,
                            receipt,
                            continuation,
                        })
                        .await?;
                        let capture = host.state.pending_capture.take();
                        let control = if let Some(capture) = capture {
                            host.complete_capture(capture, answer).await?
                        } else {
                            let completion = kcode_agent_runtime::RoundCompletion {
                                answer,
                                used_tool: host.state.used_tool,
                                finish_requested: host.state.finish_requested,
                                emitted_response: host.state.emitted_response,
                            };
                            host.complete_round(completion).await?
                        };
                        host.state.used_tool = false;
                        host.state.finish_requested = false;
                        host.state.emitted_response = false;
                        match control {
                            kcode_agent_runtime::SessionControl::Continue => {
                                Ok(TurnBoundary::Yield(turn))
                            }
                            kcode_agent_runtime::SessionControl::Complete(answer) => {
                                self.finish_stepped_turn(turn, answer, checkpoint).await
                            }
                        }
                    }
                }
            }
        }
    }

    pub async fn run_pending_turn<C, F>(
        &mut self,
        operation_id: Uuid,
        turn_deadline: Option<TurnDeadline>,
        mut checkpoint: C,
    ) -> anyhow::Result<Option<String>>
    where
        C: FnMut(Value) -> F + Send,
        F: Future<Output = anyhow::Result<()>> + Send,
    {
        let Some(turn) = self.begin_pending_turn(operation_id, turn_deadline)? else {
            return Ok(None);
        };
        let mut boundary = self.advance_pending_turn(turn, &mut checkpoint).await?;
        loop {
            boundary = match boundary {
                TurnBoundary::Await(pending) => {
                    let wake = pending.wait().await;
                    self.apply_inference_wake(wake, &mut checkpoint).await?
                }
                TurnBoundary::Yield(turn) => {
                    self.advance_pending_turn(turn, &mut checkpoint).await?
                }
                TurnBoundary::Complete(answer) => return Ok(answer),
            };
        }
    }

    fn project_descendant<T>(
        &mut self,
        outcome: Result<kcode_intelligence_router::Accounted<T>, services::ApiError>,
    ) -> anyhow::Result<T> {
        match outcome {
            Ok(accounted) => {
                kcode_intelligence_chatend::record_descendant_receipt(
                    &mut self.journal,
                    &accounted.receipt,
                )?;
                Ok(accounted.value)
            }
            Err(error) => {
                if let Some(receipt) = &error.receipt {
                    kcode_intelligence_chatend::record_descendant_receipt(
                        &mut self.journal,
                        receipt,
                    )?;
                }
                Err(error.into())
            }
        }
    }

    async fn run_subagent(
        &mut self,
        model: String,
        reasoning_effort: Option<String>,
        context_node_ids: Vec<String>,
        task: String,
        parent_operation_id: Uuid,
    ) -> anyhow::Result<String> {
        let reasoning_effort =
            reasoning_effort.unwrap_or_else(|| self.runtime.reasoning_effort.clone());
        let mut selected_node_descriptions = Vec::with_capacity(context_node_ids.len());
        for node_id in &context_node_ids {
            selected_node_descriptions.push(self.api.kmap_node(node_id)?.data.long_description);
        }
        let user_id = self
            .root_node_ids
            .first()
            .context("session has no user root for subagent intelligence accounting")?
            .clone();
        let timeout = self.agent_request_timeout();
        let runtime = self.api.agent_runtime();
        let provider = runtime.resolve_model(&model).await?.provider;
        let first_event = self.journal.state().events.len();
        let cost_before = self.projection().status;
        let subagent_context = SubagentContext::new(
            self.root_node_ids.clone(),
            self.api.loads_fixed_connections(),
            provider,
            self.subagent_codex_prompt.clone(),
            selected_node_descriptions,
        )?;
        let initial_sections = subagent_context.initial_sections().to_vec();
        let result = {
            let mut host = KennedySubagentHost {
                session: self,
                context: subagent_context,
                captures: HashMap::new(),
            };
            runtime
                .run(
                    kcode_agent_runtime::RunRequest {
                        user_id,
                        parent_operation_id,
                        model,
                        reasoning_effort,
                        context: initial_sections,
                        task,
                        timeout,
                        start_metadata: json!({"contextNodeIds":context_node_ids}),
                    },
                    &mut host,
                )
                .await
        };
        match result {
            Ok(result) => {
                let cost_after = self.projection().status;
                Ok(format!(
                    "{}\n\n[{}]",
                    result.answer,
                    cost_summary(
                        "subagent cost",
                        cost_after
                            .estimated_cost_usd_nanos
                            .saturating_sub(cost_before.estimated_cost_usd_nanos),
                        cost_after
                            .unpriced_provider_calls
                            .saturating_sub(cost_before.unpriced_provider_calls),
                    )
                ))
            }
            Err(error) => {
                let may_have_effects =
                    self.journal.state().events[first_event..]
                        .iter()
                        .any(|event| {
                            matches!(
                                &event.kind,
                                EventKind::Note { label, .. } if label == "subagent_tool_call"
                            )
                        });
                if may_have_effects {
                    Err(error.context(
                        "the subagent failed after making Ktool calls; some tool effects may already have occurred",
                    ))
                } else {
                    Err(error)
                }
            }
        }
    }

    async fn complete_subagent_freeform_write(
        &mut self,
        context: &mut SubagentContext,
        request: FreeformWrite,
        contents: String,
        budget: &kcode_agent_runtime::ContextBudget,
    ) -> anyhow::Result<kcode_agent_runtime::ToolOutcome> {
        let kind = request.kind();
        let freeform_tool = request.write_tool();
        anyhow::ensure!(
            context.source_is_open(kind, request.name()),
            "{} {:?} is not open in this subagent context. Call {} first.",
            kind.label(),
            request.name(),
            kind.open_tool()
        );
        let backend_arguments = request.capture_subagent(&mut self.journal, &now(), contents)?;
        let preview = self
            .api
            .managed_source_execute(
                &self.rust_lib_session_id,
                request.preview_tool(),
                backend_arguments.clone(),
                Vec::new(),
            )
            .await?;
        let preview = preview
            .snapshot
            .context("subagent freeform write preview omitted its source snapshot")?;
        let preview_state = context.source_state(&preview);
        anyhow::ensure!(
            budget.fits_state(preview_state.key, preview_state.text),
            "{freeform_tool} was not run because its resulting source state would exceed the subagent context limit"
        );
        let execution = self
            .api
            .managed_source_execute(
                &self.rust_lib_session_id,
                freeform_tool,
                backend_arguments,
                Vec::new(),
            )
            .await?;
        let snapshot = execution
            .snapshot
            .context("subagent freeform write omitted its resulting source snapshot")?;
        let state = context.apply_source_snapshot(snapshot);
        Ok(kcode_agent_runtime::ToolOutcome {
            text: execution.text,
            ok: true,
            state_updates: state.update.into_iter().collect(),
            displayed_state_keys: Vec::new(),
            capture: None,
        })
    }

    async fn complete_freeform_write(
        &mut self,
        pending: PendingFreeformWrite,
        contents: String,
    ) -> anyhow::Result<ToolOutcome> {
        let request = pending.request;
        let freeform_tool = request.write_tool();
        let backend_arguments =
            request.capture(&mut self.journal, &now(), pending.call_box_id, contents)?;
        let preview_result = self
            .api
            .managed_source_execute(
                &self.rust_lib_session_id,
                request.preview_tool(),
                backend_arguments.clone(),
                Vec::new(),
            )
            .await;
        let preview = match preview_result {
            Ok(preview) => preview,
            Err(error) => {
                return Ok(ToolOutcome {
                    text: format!("{freeform_tool} failed: {error}"),
                    store_result: true,
                    ok: false,
                    end_session: false,
                    freeform_write: None,
                    managed_source_snapshot: None,
                    exact_result: false,
                });
            }
        };
        let _preview = preview
            .snapshot
            .context("freeform write preview omitted the resulting source snapshot")?;
        request.source_box_id(&self.journal)?;

        let execution_result = self
            .api
            .managed_source_execute(
                &self.rust_lib_session_id,
                freeform_tool,
                backend_arguments,
                Vec::new(),
            )
            .await;
        let execution = match execution_result {
            Ok(execution) => execution,
            Err(error) => {
                return Ok(ToolOutcome {
                    text: format!("{freeform_tool} failed: {error}"),
                    store_result: true,
                    ok: false,
                    end_session: false,
                    freeform_write: None,
                    managed_source_snapshot: None,
                    exact_result: false,
                });
            }
        };
        let snapshot = execution
            .snapshot
            .context("freeform write omitted the resulting source snapshot")?;
        apply_snapshot(&mut self.journal, &now(), snapshot)?;
        Ok(ToolOutcome {
            text: execution.text,
            store_result: false,
            ok: true,
            end_session: false,
            freeform_write: None,
            managed_source_snapshot: None,
            exact_result: false,
        })
    }

    async fn send_telegram_dm(&mut self, arguments: &Value) -> anyhow::Result<String> {
        let request = kcode_telegram_session_coordinator::parse_private_request(arguments)?;
        let attachments = self.telegram_delivery_attachments(request.attachments)?;
        let caller_holds_user_lock = self.session_type == "telegram"
            && self.channel.get("telegramUserId").and_then(Value::as_i64)
                == Some(request.telegram_user_id);
        self.api
            .telegram()
            .send_private(kcode_telegram_session_coordinator::PrivateDelivery {
                telegram_user_id: request.telegram_user_id,
                message: request.message,
                attachments,
                caller_holds_user_lock,
            })
            .await
    }

    async fn send_telegram_group_message(&mut self, arguments: &Value) -> anyhow::Result<String> {
        let request = kcode_telegram_session_coordinator::parse_group_request(arguments)?;
        let attachments = self.telegram_delivery_attachments(request.attachments)?;
        self.api
            .telegram()
            .send_group(kcode_telegram_session_coordinator::GroupDelivery {
                root_node_id: request.root_node_id,
                message: request.message,
                attachments,
            })
            .await
    }

    fn telegram_delivery_attachments(
        &mut self,
        requests: Vec<kcode_telegram_session_coordinator::AttachmentRequest>,
    ) -> anyhow::Result<Vec<kcode_telegram_session_coordinator::Attachment>> {
        let api = self.api.clone();
        kcode_kennedy_session_objects::delivery_attachments(
            &mut self.journal,
            requests,
            move |canonical_id| api.kmap_file(canonical_id).map_err(Into::into),
        )
    }

    async fn execute_tool(
        &mut self,
        call: &ToolCall,
        operation_id: Uuid,
    ) -> anyhow::Result<ToolOutcome> {
        self.assert_tool_allowed(&call.name)?;
        anyhow::ensure!(
            call.name != LAUNCH_SESSION_TOOL,
            "LaunchSession requires the checkpointed launch dispatch lane"
        );
        let decoded = decode(&call.name, &call.arguments)?;
        let mut end_session = false;
        let mut store_result = true;
        let mut freeform_write = None;
        let mut managed_source_snapshot = None;
        let text = match (call.name.as_str(), decoded) {
            ("NoteToSelf", None) => {
                decode_note_to_self(&call.arguments)?;
                store_result = false;
                "Note saved.".into()
            }
            ("SendTelegramDM", _) => self.send_telegram_dm(&call.arguments).await?,
            ("SendTelegramGroupMessage", _) => {
                self.send_telegram_group_message(&call.arguments).await?
            }
            (
                "RunSubagent",
                Some(DecodedTool::RunSubagent {
                    model,
                    reasoning_effort,
                    context_node_ids,
                    task,
                }),
            ) => {
                let first_event = self.journal.state().events.len();
                match self
                    .run_subagent(
                        model,
                        reasoning_effort,
                        context_node_ids,
                        task,
                        operation_id,
                    )
                    .await
                {
                    Ok(response) => response,
                    Err(error) => {
                        let may_have_effects = self.journal.state().events[first_event..]
                            .iter()
                            .any(|event| {
                                matches!(
                                    &event.kind,
                                    EventKind::Note { label, .. }
                                        if label == "subagent_tool_call"
                                )
                            });
                        if may_have_effects {
                            return Err(error.context(
                                "the subagent failed after making Ktool calls; some tool effects may already have occurred",
                            ));
                        }
                        return Err(error);
                    }
                }
            }
            ("EndSession", Some(DecodedTool::EndSession { message })) => {
                anyhow::ensure!(
                    !matches!(self.mode, AgentMode::Conversation),
                    "EndSession is only available during an autonomous or history-ingress session"
                );
                end_session = true;
                if matches!(self.mode, AgentMode::FreeTime)
                    && let Some(message) = message.filter(|message| !message.trim().is_empty())
                {
                    self.free_time["nextSessionMessage"] = json!(message);
                }
                "Session ending.".into()
            }
            ("DehydrateBoxes", Some(DecodedTool::BoxIds(ids))) => {
                self.journal.dehydrate_boxes(now(), &ids)?;
                format!(
                    "Dehydrated boxes {}.",
                    ids.iter()
                        .map(ToString::to_string)
                        .collect::<Vec<_>>()
                        .join(", ")
                )
            }
            ("SummarizeBox", Some(DecodedTool::SummarizeBox { box_id, summary })) => {
                self.journal.summarize_box(now(), box_id, summary)?;
                format!("Summarized box {box_id}.")
            }
            ("HydrateBox", Some(DecodedTool::BoxId(id))) => {
                self.journal.rehydrate_box(now(), id)?;
                let external_event_id = self.pending_external_event_id.clone();
                match self.recover_context_overflow(external_event_id.as_deref(), &[id])? {
                    ContextRecovery::NotNeeded => format!("Hydrated box {id}."),
                    ContextRecovery::Recovered => {
                        format!("Hydrated box {id}.\n\n{CONTEXT_OVERFLOW_WARNING}")
                    }
                    ContextRecovery::Irreducible => anyhow::bail!(CONTEXT_OVERFLOW_WARNING),
                }
            }
            ("BoxesIntoObjects", Some(DecodedTool::BoxIds(ids))) => {
                kcode_kennedy_box_text_objects::stage_box_text_objects(
                    &mut self.journal,
                    &ids,
                    &now(),
                )?
            }
            ("LoadNodes", Some(DecodedTool::LoadNodes(identifiers))) => {
                load_durable_batch(self.api.kmap(), &mut self.context, &identifiers)?;
                let changed = self.sync_kweb_boxes()?;
                store_result = false;
                render_load_nodes_result(
                    &self.journal,
                    &changed,
                    &self.runtime_budget().footer_lines(),
                )?
            }
            (
                "EmitObject",
                Some(DecodedTool::EmitObject {
                    object_id,
                    file_name,
                }),
            ) => {
                anyhow::ensure!(
                    matches!(self.mode, AgentMode::Conversation),
                    "EmitObject is only available in a conversation"
                );
                let object = self.resolve_object(&object_id)?;
                let file_name = file_name.unwrap_or_else(|| object.file_name.clone());
                if let Some(maximum) = self.channel.get("maxObjectBytes").and_then(Value::as_u64) {
                    anyhow::ensure!(
                        !object.bytes.is_empty(),
                        "object {object_id} is empty and cannot be sent through this channel"
                    );
                    anyhow::ensure!(
                        object.bytes.len() as u64 <= maximum,
                        "object {object_id} is {} bytes, over this channel's {maximum}-byte limit",
                        object.bytes.len()
                    );
                }
                let descriptor = json!({
                    "objectId":object_id,
                    "fileName":file_name,
                    "mediaType":object.media_type,
                    "byteLength":object.bytes.len(),
                });
                let mut metadata = json!({
                    "outputKind":"object",
                    "attachments":[descriptor.clone()],
                });
                if let Some(external_event_id) = &self.pending_external_event_id {
                    metadata["externalEventId"] = json!(external_event_id);
                }
                let content = BoxContent {
                    text: String::new(),
                    objects: vec![object_id.clone()],
                    metadata,
                };
                self.journal
                    .create_box(now(), "Kennedy message", BoxOwner::Kennedy, content)?;
                let mut transcript = json!({
                    "role":"kennedy",
                    "content":"",
                    "objects":[object_id],
                    "attachments":[descriptor],
                });
                if let Some(external_event_id) = &self.pending_external_event_id {
                    transcript["externalEventId"] = json!(external_event_id);
                }
                self.transcript.push(transcript);
                store_result = false;
                "Object emitted to the user.".into()
            }
            ("WebSearch", Some(DecodedTool::WebSearch { question, model })) => {
                let user_id = self
                    .root_node_ids
                    .first()
                    .context("session has no user root for intelligence accounting")?
                    .clone();
                let outcome = self
                    .api
                    .search(
                        &user_id,
                        kcode_intelligence_router::SearchRequest {
                            question,
                            model,
                            operation_id: Uuid::new_v4(),
                            parent_operation_id: Some(operation_id),
                        },
                    )
                    .await;
                let result = self.project_descendant(outcome)?;
                render_web_search_result(&result)?
            }
            ("WebFetch", Some(DecodedTool::WebFetch(url))) => {
                let user_id = self
                    .root_node_ids
                    .first()
                    .context("session has no user root for intelligence accounting")?;
                let result = self
                    .api
                    .fetch(
                        user_id,
                        kcode_intelligence_router::FetchRequest {
                            url,
                            operation_id: Uuid::new_v4(),
                            parent_operation_id: Some(operation_id),
                        },
                    )
                    .await?;
                render_web_fetch_result(&result)?
            }
            ("StageTelegramGroupMedia", Some(DecodedTool::StageTelegramGroupMedia(message_id))) => {
                let media_ref = kcode_telegram_session_coordinator::group_media_reference(
                    &self.group_context,
                    message_id,
                )?;
                let chat_id = media_ref.chat_id;
                let api = self.api.clone();
                let staged = kcode_kennedy_session_objects::stage_telegram_group_media(
                    &mut self.journal,
                    kcode_kennedy_session_objects::TelegramStageRequest {
                        chat_id,
                        message_id,
                        maximum_bytes: MAX_MEDIA_ENRICHMENT_BYTES,
                        transport_metadata: media_ref.transport_metadata(),
                        recorded_at: now(),
                    },
                    || api.telegram().group_message_media(chat_id, message_id),
                    |media_type| {
                        kcode_telegram_session_coordinator::group_media_file_name(
                            &media_ref, media_type,
                        )
                    },
                )?;
                render(RenderRequest::StagedTelegramMedia {
                    pending_id: &staged.descriptor.pending_id,
                    kind: &staged.kind,
                    file_name: &staged.descriptor.file_name,
                    media_type: &staged.descriptor.media_type,
                    size_bytes: staged.descriptor.size_bytes,
                    message_id,
                    reused: staged.reused,
                })?
            }
            (
                "TranscribeAudio",
                Some(DecodedTool::MediaEnrichment {
                    object_id,
                    model,
                    prompt,
                }),
            ) => {
                let object = self.resolve_media_object(&object_id)?;
                validate(ValidationRequest::TranscribableAudio(&object.media_type))?;
                validate(ValidationRequest::TranscriptionModel(&model))?;
                let user_id = self
                    .root_node_ids
                    .first()
                    .context("session has no user root for intelligence accounting")?
                    .clone();
                let outcome = self
                    .api
                    .transcribe_audio(
                        &user_id,
                        &model,
                        &prompt,
                        object.bytes,
                        object.file_name.clone(),
                        &object.media_type,
                        None,
                        operation_id,
                    )
                    .await;
                let result = self.project_descendant(outcome)?;
                render_audio_transcription_result(
                    &object.object_id,
                    &object.file_name,
                    &object.media_type,
                    &result,
                )?
            }
            (
                "AnnotateMedia",
                Some(DecodedTool::MediaEnrichment {
                    object_id,
                    model,
                    prompt,
                }),
            ) => {
                let media = self.resolve_media_object(&object_id)?;
                validate(ValidationRequest::Annotation {
                    model: &model,
                    media_type: &media.media_type,
                })?;
                let user_id = self
                    .root_node_ids
                    .first()
                    .context("session has no user root for intelligence accounting")?
                    .clone();
                let outcome = self
                    .api
                    .annotate_media(
                        &user_id,
                        &model,
                        &prompt,
                        media.bytes,
                        media.file_name.clone(),
                        &media.media_type,
                        operation_id,
                    )
                    .await;
                let result = self.project_descendant(outcome)?;
                render_media_annotation_result(
                    &media.object_id,
                    &media.file_name,
                    &media.media_type,
                    &result,
                )?
            }
            (
                "GenerateImage",
                Some(DecodedTool::GenerateImage {
                    model,
                    prompt,
                    reference_object_ids,
                }),
            ) => {
                let mut references = Vec::with_capacity(reference_object_ids.len());
                for object_id in &reference_object_ids {
                    references.push(self.resolve_image_object(object_id)?);
                }
                let user_id = self
                    .root_node_ids
                    .first()
                    .context("session has no user root for intelligence accounting")?
                    .clone();
                let outcome = self
                    .api
                    .generate_image(&user_id, &model, &prompt, references, operation_id)
                    .await;
                let result = self.project_descendant(outcome)?;
                let size = result.bytes.len();
                let file_name =
                    format!("generated-image.{}", image_extension(&result.content_type));
                let object_id = self.api.save_generated_image(
                    result.bytes,
                    &file_name,
                    &result.content_type,
                    &result.model,
                )?;
                format!(
                    "Generated image.\nObject: {object_id}\nFile: {file_name}\nContent type: {}\nSize: {size} bytes\nModel: {}\nUse EmitObject with {object_id} to deliver it.",
                    result.content_type, result.model
                )
            }
            ("ExtractDocumentText", Some(DecodedTool::ObjectId(object_id))) => {
                let object = self.resolve_media_object(&object_id)?;
                validate(ValidationRequest::ExtractableDocument {
                    media_type: &object.media_type,
                    file_name: &object.file_name,
                })?;
                let result = self
                    .api
                    .extract_document(object.bytes, object.file_name.clone(), &object.media_type)
                    .await?;
                render_document_extraction_result(&object.object_id, &object.file_name, &result)?
            }
            (name, None) if SPEECH_CLASSIFICATION_TOOLS.contains(&name) => {
                self.api
                    .execute_speech_classification_tool(name, call.arguments.clone())
                    .await?
            }
            (name, None) if TASK_BOARD_TOOLS.contains(&name) => {
                self.execute_task_board_tool(name, &call.arguments).await?
            }
            (name, Some(decoded)) if is_kweb_mutation(name) => {
                let (text, _) = execute_kweb_mutation(
                    name,
                    decoded,
                    &self.context,
                    &mut self.plan,
                    &mut self.journal,
                )?;
                self.sync_kweb_boxes()?;
                text
            }
            (name, None)
                if RUST_LIB_TOOLS.contains(&name)
                    || WEB_LIB_TOOLS.contains(&name)
                    || RUST_BIN_TOOLS.contains(&name) =>
            {
                if let Some(request) = prepare_freeform_write(&self.journal, name, &call.arguments)?
                {
                    store_result = false;
                    let acknowledgement = request.acknowledgement();
                    freeform_write = Some(request);
                    acknowledgement
                } else {
                    let object_ids = if name == CALL_RUST_BIN_TOOL {
                        decode_managed_objects(ManagedObjectArguments::RustBinary(&call.arguments))?
                    } else if name == ATTACH_OBJECT_WEB_LIB_TOOL {
                        decode_managed_objects(ManagedObjectArguments::WebLibraryAttachment(
                            &call.arguments,
                        ))?
                    } else {
                        Vec::new()
                    };
                    let mut objects = Vec::with_capacity(object_ids.len());
                    for object_id in object_ids {
                        objects.push(self.resolve_object(&object_id)?.bytes);
                    }
                    let execution = self
                        .api
                        .managed_source_execute(
                            &self.rust_lib_session_id,
                            name,
                            call.arguments.clone(),
                            objects,
                        )
                        .await?;
                    if let Some(snapshot) = execution.snapshot {
                        managed_source_snapshot = Some(snapshot);
                        store_result = false;
                    }
                    execution.text
                }
            }
            (name, Some(_)) => {
                anyhow::bail!("decoded contract for {name} did not match its dispatch lane")
            }
            (name, None) => anyhow::bail!("Tool {name} is not available"),
        };
        Ok(ToolOutcome {
            text,
            store_result,
            ok: true,
            end_session,
            freeform_write,
            managed_source_snapshot,
            exact_result: false,
        })
    }

    async fn execute_task_board_tool(
        &self,
        name: &str,
        arguments: &Value,
    ) -> anyhow::Result<String> {
        let board = self
            .api
            .task_board()
            .context("task board is not configured")?
            .clone();
        let name = name.to_owned();
        let arguments = arguments.clone();
        let user_id = self
            .root_node_ids
            .first()
            .context("session has no user root for task-category lookup")?
            .clone();
        tokio::task::spawn_blocking(move || -> anyhow::Result<String> {
            let output = match name.as_str() {
                "CreateTaskCategory" => serde_json::to_string_pretty(
                    &board.create_category(serde_json::from_value(arguments)?)?,
                )?,
                "GetTaskCategory" => {
                    let call: CategoryCall = serde_json::from_value(arguments)?;
                    serde_json::to_string_pretty(&board.category(
                        &call.category_id,
                        kcode_task_board::BrowsePage {
                            user_id,
                            offset: call.offset,
                            limit: call.limit,
                        },
                    )?)?
                }
                "RemoveTaskCategory" => {
                    let call: CategoryId = serde_json::from_value(arguments)?;
                    board.remove_category(&call.category_id)?;
                    format!("Removed category {}.", call.category_id)
                }
                "CreateTask" => serde_json::to_string_pretty(
                    &board.create_task(serde_json::from_value(arguments)?)?,
                )?,
                "GetTask" => {
                    let call: TaskId = serde_json::from_value(arguments)?;
                    serde_json::to_string_pretty(&board.task(&call.task_id)?)?
                }
                "UpdateTask" => serde_json::to_string_pretty(
                    &board.update_task(serde_json::from_value(arguments)?)?,
                )?,
                "RemoveTask" => {
                    let call: TaskId = serde_json::from_value(arguments)?;
                    board.remove_task(&call.task_id)?;
                    format!("Removed task {}.", call.task_id)
                }
                "GetTopTaskOrphan" => {
                    let _: EmptyCall = serde_json::from_value(arguments)?;
                    serde_json::to_string_pretty(&board.top_orphan()?)?
                }
                _ => anyhow::bail!("Tool {name} is not a task-board operation"),
            };
            Ok(output)
        })
        .await
        .context("task-board worker stopped")?
    }

    fn assert_tool_allowed(&self, name: &str) -> anyhow::Result<()> {
        let write = matches!(
            name,
            "ConnectNodes"
                | "ConsolidateFanout"
                | "SetFixedConnection"
                | "CreateNode"
                | "UpdateNode"
        );
        anyhow::ensure!(
            !write || !matches!(self.mode, AgentMode::Conversation),
            "{name} requires the global Kweb write lane and is unavailable in a read-only conversation"
        );
        if name == "EndSession" {
            anyhow::ensure!(
                !matches!(self.mode, AgentMode::Conversation),
                "EndSession is unavailable in a conversation"
            );
        }
        if name == LAUNCH_SESSION_TOOL {
            anyhow::ensure!(
                self.launch_session_authorized(),
                "LaunchSession is unavailable without a genuine current user turn in an eligible conversation"
            );
        }
        Ok(())
    }

    fn sync_kweb_boxes(&mut self) -> anyhow::Result<Vec<BoxId>> {
        let (updates, creates) = self.plan.context_projection();
        self.context
            .sync_chatend(&mut self.journal, now(), &updates, &creates)
            .map_err(anyhow::Error::new)
    }

    fn record_tool_invocation(
        &mut self,
        name: &str,
        arguments: Value,
    ) -> anyhow::Result<RecordedToolInvocation> {
        let invocation_id = Uuid::new_v4().to_string();
        let invocation = RecordedToolInvocation {
            tool_instance: tool_instance_for_invocation(name, &invocation_id),
            invocation_id,
            tool_name: name.into(),
        };
        self.journal.record(
            now(),
            EventKind::ToolInvoked {
                tool_instance: invocation.tool_instance.clone(),
                tool_name: invocation.tool_name.clone(),
                arguments,
                invocation_id: Some(invocation.invocation_id.clone()),
            },
        )?;
        Ok(invocation)
    }

    fn record_tool_completion(
        &mut self,
        invocation: Option<&RecordedToolInvocation>,
        outcome: Value,
    ) -> anyhow::Result<EventId> {
        record_tool_completion_event(&mut self.journal, invocation, outcome)
    }

    fn finalize_kweb_session(&mut self) -> anyhow::Result<()> {
        self.provider_affinity = None;
        self.next_thread_reset_reason = None;
        if self.commit_receipt.is_some() {
            return Ok(());
        }
        self.repair_unfinished_tools()?;
        self.journal.seal()?;
        let archive = self.journal.archive_bytes()?;
        let object_locations = self
            .journal
            .objects()
            .iter()
            .map(|(id, location)| (id.clone(), location.clone()))
            .collect::<Vec<_>>();
        let mut objects = BTreeMap::new();
        for (id, location) in object_locations {
            let pending_id = id.to_string();
            let transport_kind =
                kcode_kennedy_session_objects::staged_descriptor(&self.journal, &id)?
                    .transport_kind;
            let bytes = encode_file(
                &pending_id,
                location.metadata.file_name.as_deref(),
                &location.metadata.media_type,
                transport_kind.as_deref(),
                self.journal.read_object(&id)?,
            )
            .with_context(|| format!("encoding staged object {pending_id}"))?;
            anyhow::ensure!(
                objects.insert(pending_id.clone(), bytes).is_none(),
                "duplicate staged object {pending_id}"
            );
        }
        let material = self.plan.commit_material()?;
        let result = self.api.commit_kweb_session(CommitRequest {
            idempotency_key: self.journal.state().metadata.session_id.clone(),
            author: self.commit_author.clone(),
            source_created_at: DateTime::parse_from_rfc3339(&self.started_at)
                .context("session start timestamp is invalid")?
                .with_timezone(&Utc),
            archive,
            objects,
            creates: material.creates,
            updates: material.updates,
        })?;
        self.journal
            .mark_completed(result.session_object_id.to_string());
        self.commit_receipt = Some(result);
        Ok(())
    }

    fn prepare_free_time_round(&mut self) -> anyhow::Result<bool> {
        if !matches!(self.mode, AgentMode::FreeTime) {
            return Ok(false);
        }
        let Some(deadline) = deadline(&self.free_time) else {
            return Ok(false);
        };
        if Utc::now() >= deadline {
            self.free_time_end_reason = Some("deadline".into());
            self.journal.create_box(
                now(),
                "Self-time timer",
                BoxOwner::Controller,
                BoxContent::text(
                    "The self-time deadline has arrived. Finish without starting more tool work.",
                ),
            )?;
            return Ok(true);
        }
        Ok(false)
    }

    fn agent_request_timeout(&self) -> Option<Duration> {
        if matches!(self.mode, AgentMode::Conversation) && self.session_type == "conversation" {
            return Some(BROWSER_CONVERSATION_REQUEST_TIMEOUT);
        }
        if matches!(self.mode, AgentMode::Ingress { .. }) {
            return Some(HISTORY_INGRESS_REQUEST_TIMEOUT);
        }
        if matches!(self.mode, AgentMode::Wakeup) {
            return Some(WAKEUP_REQUEST_TIMEOUT);
        }
        if matches!(self.mode, AgentMode::FreeTime) {
            let deadline = deadline(&self.free_time)?;
            return Some(Duration::from_secs(
                (deadline - Utc::now()).num_seconds().max(1) as u64
                    + SELF_TIME_HARD_STOP_ALLOWANCE.as_secs(),
            ));
        }
        None
    }

    pub fn refresh_telegram_group_context(
        &mut self,
        group_context: &Value,
        current_message_id: Option<&str>,
    ) -> anyhow::Result<()> {
        if self.session_type != "telegram-group" {
            return Ok(());
        }
        self.invalidate_active_stepped_turn();
        self.channel["groupContext"] = group_context.clone();
        self.group_context = group_context.clone();
        self.journal.create_box(
            now(),
            "Telegram group update",
            BoxOwner::Controller,
            BoxContent::text(kcode_telegram_session_coordinator::format_group_context(
                group_context,
            )),
        )?;
        self.recover_context_overflow(current_message_id, &[])?;
        Ok(())
    }

    pub fn finalize_free_time(&mut self, reason: &str) -> anyhow::Result<()> {
        anyhow::ensure!(
            matches!(reason, "tool" | "deadline" | "hard-stop" | "user-stop"),
            "invalid self-time completion reason"
        );
        self.invalidate_active_stepped_turn();
        self.free_time["sliceEndedReason"] = json!(reason);
        self.free_time["sliceEndedAt"] = json!(now());
        self.pending_turn = false;
        self.pending_external_event_id = None;
        self.clear_launch_turn_authority();
        Ok(())
    }

    pub fn commit_current_write_session(&mut self) -> anyhow::Result<()> {
        anyhow::ensure!(
            matches!(
                self.mode,
                AgentMode::FreeTime | AgentMode::Wakeup | AgentMode::Ingress { .. }
            ),
            "a read-only conversation cannot be committed as a Kweb write session"
        );
        self.invalidate_active_stepped_turn();
        self.finalize_kweb_session()?;
        self.completed = true;
        Ok(())
    }

    pub fn snapshot(&self) -> anyhow::Result<Value> {
        let projection = self.projection();
        let submitted = self
            .journal
            .state()
            .current_ingress_attempt_events()
            .iter()
            .rev()
            .find_map(|event| {
                let EventKind::ProviderInputSubmitted { round, context, .. } = &event.kind else {
                    return None;
                };
                Some((event.recorded_at.as_str(), *round, context))
            });
        let (chatend_text, chatend_text_source, structured_material) = match submitted {
            Some((submitted_at, round, submitted)) => (
                submitted.input.clone(),
                "submitted",
                json!({
                    "provider":submitted.provider,
                    "model":submitted.model,
                    "reasoningEffort":submitted.reasoning_effort,
                    "baseInstructions":submitted.base_instructions,
                    "developerInstructions":submitted.developer_instructions,
                    "tools":submitted.tools,
                    "round":round,
                    "submittedAt":submitted_at,
                }),
            ),
            None => (projection.render(), "reconstructed", Value::Null),
        };
        let session_status = projection.status.clone();
        let completed_invocations = completed_invocation_ids(&self.journal);
        let launch_intents = pruned_launch_intents(
            &self.launch_intents,
            self.launch_user_turn_id,
            &completed_invocations,
        );
        Ok(json!({
            "format":"kennedy-chatend",
            "version":1,
            "stateVersion":CHECKPOINT_STATE_VERSION,
            "sessionId":self.journal.state().metadata.session_id,
            "chatendMetadata":self.journal.state().metadata,
            "sessionType":self.session_type,
            "sourceSessionType":self.source_session_type,
            "channel":self.channel,
            "freeTime":self.free_time,
            "orchestration":self.orchestration,
            "provenanceId":self.provenance_id,
            "launchProvenance":self.launch_provenance,
            "launchContextNodeIds":self.launch_context_node_ids,
            "launchUserTurnId":self.launch_user_turn_id,
            "launchIntents":launch_intents,
            "rustLibSessionId":self.rust_lib_session_id,
            "rootNodeIds":self.root_node_ids,
            "referenceRootNodeIds":self.reference_root_node_ids,
            "startedAt":self.started_at,
            "transcript":self.transcript,
            "pendingTurn":self.pending_turn,
            "pendingExternalEventId":self.pending_external_event_id,
            "roundsUsed":self.rounds_used,
            "providerAffinity":self.provider_affinity,
            "nextThreadResetReason":self.next_thread_reset_reason,
            "completed":self.completed,
            "sessionObjectId":self.journal.state().completed_session_object,
            "commitReceipt":self.commit_receipt,
            "commitAuthor":self.commit_author,
            "providerModel":self.runtime.model,
            "kwebPlan":self.plan.checkpoint_value()?,
            "boxCount":self.journal.state().boxes.len(),
            "eventCount":self.journal.state().events.len(),
            "boxes":self.journal.state().boxes,
            "events":self.journal.state().events,
            "context":projection,
            "sessionStatus":session_status,
            "chatendText":chatend_text,
            "chatendTextSource":chatend_text_source,
            "structuredMaterial":structured_material,
        }))
    }

    pub async fn release_managed_sources(&self) {
        self.api
            .release_managed_sources(&self.rust_lib_session_id)
            .await;
    }
}

impl<C, F> kcode_agent_runtime::SessionHost for KennedySessionHost<'_, C>
where
    C: FnMut(Value) -> F + Send,
    F: Future<Output = anyhow::Result<()>> + Send,
{
    fn prepare_round<'a>(
        &'a mut self,
        round: u64,
    ) -> kcode_agent_runtime::HostFuture<'a, kcode_agent_runtime::RoundPreparation> {
        Box::pin(async move {
            self.session.rounds_used = round;
            self.state.deadline_after_response = self.session.prepare_free_time_round()?;
            let external_event_id = self.session.pending_external_event_id.clone();
            if self
                .session
                .recover_context_overflow(external_event_id.as_deref(), &[])?
                == ContextRecovery::Irreducible
                || (matches!(self.session.mode, AgentMode::Ingress { .. })
                    && self.session.ingress_force_commit_requested())
            {
                return Ok(kcode_agent_runtime::RoundPreparation::Complete(None));
            }
            let ingress_time_remaining = self.session.ingress_time_remaining()?;
            let timeout = self.session.agent_request_timeout();
            self.session.begin_provider_call_budget(timeout);
            let tool_description = call_ktool_description(self.session.launch_session_authorized());
            let material_fingerprint = self
                .session
                .provider_material_fingerprint(&tool_description);
            let mut thread_reset_reason = self.session.next_thread_reset_reason.take();
            let mut continuation = None;
            let mut resume_after = None;
            if let Some(affinity) = &self.session.provider_affinity {
                if affinity.material_fingerprint == material_fingerprint {
                    continuation = Some(affinity.continuation.clone());
                    resume_after = Some(affinity.synchronized_event_id);
                } else {
                    self.session.provider_affinity = None;
                    thread_reset_reason = Some("provider_material_changed".into());
                }
            }
            if continuation.is_some() {
                self.session.provider_affinity = None;
                self.session.next_thread_reset_reason =
                    Some("prior_provider_turn_ambiguous".into());
            }
            let footer_lines = self.session.runtime_budget().footer_lines();
            let prepared = if let Some(remaining_seconds) = ingress_time_remaining {
                self.session
                    .journal
                    .prepare_provider_projection_with_ingress_time(
                        now(),
                        &footer_lines,
                        &material_fingerprint,
                        resume_after,
                        remaining_seconds,
                        self.session.previous_ingress_attempt_timed_out,
                    )?
            } else {
                self.session.journal.prepare_provider_projection(
                    now(),
                    &footer_lines,
                    &material_fingerprint,
                    resume_after,
                )?
            };
            if let Some(reason) = prepared.thread_reset_reason.clone() {
                self.session.provider_affinity = None;
                continuation = None;
                thread_reset_reason = Some(reason);
            }
            if continuation.is_none() && thread_reset_reason.is_some() {
                self.session.next_thread_reset_reason = thread_reset_reason.clone();
            }
            let input = prepared.projection.render();
            let projection_hash = hex::encode(Sha256::digest(input.as_bytes()));
            let provider_input_hash =
                hex::encode(Sha256::digest(prepared.provider_input.as_bytes()));
            let provider_input_bytes = prepared.provider_input.len() as u64;
            let thread_action = if continuation.is_some() {
                "resume"
            } else {
                "start"
            }
            .to_owned();
            self.state.prepared_cache = Some(PreparedCacheObservation {
                cacheable_prefix_bytes: prepared.cacheable_prefix_bytes,
                expectation: prepared.expectation,
                material_fingerprint,
                projection_hash,
                logical_input: input.clone(),
                provider_input_hash,
                provider_input_bytes,
                thread_action,
                thread_reset_reason,
                estimated_input_tokens: prepared.projection.estimated_tokens,
                raw_estimated_input_tokens: prepared.projection.raw_estimated_tokens,
                provider: String::new(),
                model: self.session.runtime.model.clone(),
            });
            Ok(kcode_agent_runtime::RoundPreparation::Run(
                kcode_agent_runtime::PreparedRound {
                    input,
                    provider_input: prepared.provider_input,
                    continuation,
                    model: self.session.runtime.model.clone(),
                    reasoning_effort: self.session.runtime.reasoning_effort.clone(),
                    tool_description,
                    timeout,
                },
            ))
        })
    }

    fn record<'a>(
        &'a mut self,
        event: kcode_agent_runtime::SessionEvent,
    ) -> kcode_agent_runtime::HostFuture<'a, ()> {
        Box::pin(async move {
            match event {
                kcode_agent_runtime::SessionEvent::InferenceSubmitted {
                    manifest_hash,
                    model,
                    ..
                } => {
                    let prepared =
                        self.state.prepared_cache.as_ref().context(
                            "inference was submitted before provider context preparation",
                        )?;
                    anyhow::ensure!(
                        prepared.projection_hash == manifest_hash,
                        "provider input hash changed after context preparation"
                    );
                    self.state.accounting = Some(kcode_intelligence_chatend::TopLevelCall::new(
                        manifest_hash.clone(),
                        model,
                    ));
                    self.session.journal.record(
                        now(),
                        EventKind::InferenceSubmitted {
                            manifest_hash,
                            estimated_input_tokens: prepared.estimated_input_tokens,
                            raw_estimated_input_tokens: Some(prepared.raw_estimated_input_tokens),
                        },
                    )?;
                }
                kcode_agent_runtime::SessionEvent::ProviderInput { round, context } => {
                    let prepared = self
                        .state
                        .prepared_cache
                        .as_ref()
                        .context("provider context arrived before context preparation")?;
                    anyhow::ensure!(
                        hex::encode(Sha256::digest(context.input.as_bytes()))
                            == prepared.provider_input_hash,
                        "provider submitted transport input different from the prepared continuation delta"
                    );
                    let provider = context.provider.clone();
                    let model = context.model.clone();
                    let synchronized_after = self.session.journal.record(
                        now(),
                        EventKind::ProviderInputSubmitted {
                            round,
                            context: ProviderContext {
                                input: prepared.logical_input.clone(),
                                provider: context.provider,
                                model: context.model,
                                reasoning_effort: context.reasoning_effort,
                                base_instructions: context.base_instructions,
                                developer_instructions: context.developer_instructions,
                                tools: context
                                    .tools
                                    .into_iter()
                                    .map(|tool| ProviderToolDefinition {
                                        name: tool.name,
                                        description: tool.description,
                                        input_schema: tool.input_schema,
                                    })
                                    .collect(),
                            },
                            transport_input_hash: Some(prepared.provider_input_hash.clone()),
                            transport_input_bytes: Some(prepared.provider_input_bytes),
                            thread_action: Some(prepared.thread_action.clone()),
                            thread_reset_reason: prepared.thread_reset_reason.clone(),
                            cacheable_prefix_bytes: prepared.cacheable_prefix_bytes,
                            material_fingerprint: prepared.material_fingerprint.clone(),
                            cache_expectation: prepared.expectation.label().into(),
                            planned_invalidation_reason: prepared
                                .expectation
                                .planned_reason()
                                .map(str::to_owned),
                        },
                    )?;
                    self.state.provider_synchronized_after = Some(synchronized_after);
                    if let Some(prepared) = self.state.prepared_cache.as_mut() {
                        prepared.provider = provider;
                        prepared.model = model;
                    }
                }
                kcode_agent_runtime::SessionEvent::UsageUpdated { usage, .. } => {
                    self.state
                        .accounting
                        .as_mut()
                        .context("provider usage arrived before inference submission")?
                        .usage_updated(&mut self.session.journal, &now(), &usage)?;
                }
                kcode_agent_runtime::SessionEvent::ProviderReceipt {
                    usage,
                    receipt,
                    continuation,
                    ..
                } => {
                    self.state
                        .accounting
                        .take()
                        .context("provider receipt arrived before inference submission")?
                        .completed(&mut self.session.journal, &now(), usage.as_ref())?;
                    let prepared = self
                        .state
                        .prepared_cache
                        .take()
                        .context("provider receipt arrived before context preparation")?;
                    if let Some(reason) = self.state.restart_fresh_reason.take() {
                        anyhow::ensure!(
                            continuation.is_none(),
                            "restart-fresh receipt unexpectedly retained a native continuation"
                        );
                        self.session.provider_affinity = None;
                        self.session.next_thread_reset_reason = Some(reason);
                    } else if let Some(continuation) = continuation {
                        anyhow::ensure!(
                            receipt.provider_thread_id.as_deref()
                                == Some(continuation.thread_id.as_str()),
                            "provider receipt thread differs from continuation state"
                        );
                        let synchronized_event_id = self
                            .session
                            .journal
                            .state()
                            .events
                            .last()
                            .context("provider completion did not create a journal event")?
                            .id;
                        self.session.provider_affinity = Some(ProviderAffinityState {
                            continuation,
                            synchronized_event_id,
                            material_fingerprint: prepared.material_fingerprint.clone(),
                        });
                        self.session.next_thread_reset_reason = None;
                    } else {
                        self.session.provider_affinity = None;
                        self.session.next_thread_reset_reason = Some(
                            if prepared.thread_action == "resume" {
                                "provider_thread_resume_unavailable"
                            } else {
                                "provider_continuation_unavailable"
                            }
                            .into(),
                        );
                    }
                    log_primary_thread_observation(
                        self.state.operation_id,
                        self.session.rounds_used,
                        &self.session.runtime.model,
                        &prepared,
                        receipt.provider_thread_id.as_deref(),
                        usage.as_ref().map_or(0, |usage| usage.input_tokens),
                        usage.as_ref().map_or(0, |usage| usage.cached_input_tokens),
                    );
                }
            }
            let snapshot = self.session.snapshot()?;
            (self.checkpoint)(snapshot).await
        })
    }

    fn execute_tool<'a>(
        &'a mut self,
        call: anyhow::Result<kcode_agent_runtime::ToolCall>,
        operation_id: Uuid,
    ) -> kcode_agent_runtime::HostFuture<'a, kcode_agent_runtime::SessionToolOutcome> {
        Box::pin(async move {
            if let Some(pending) = &self.state.pending_freeform_write {
                let text = format!(
                    "{} is awaiting the complete file contents; no other Ktool can run before that output.",
                    pending.request.write_tool()
                );
                self.session
                    .record_tool_completion(None, json!({"ok":false,"result":text}))?;
                return Ok(kcode_agent_runtime::SessionToolOutcome {
                    text,
                    ok: false,
                    capture: Some(json!(true)),
                    stop: false,
                    finish_after_round: false,
                    emitted_response: false,
                });
            }
            let tool_started_at = std::time::Instant::now();
            let mut created_call_box_id = None;
            let mut recorded_invocation = None;
            let transcript_start = self.session.transcript.len();
            let mut emitted_response = false;
            let mut outcome = match call {
                Ok(call) => {
                    let call = ToolCall {
                        name: call.name,
                        arguments: call.arguments,
                    };
                    let call_name = format!("Kennedy tool call: {}", call.name);
                    let call_content = tool_invocation_content(&call.name, &call.arguments)?;
                    recorded_invocation = Some(
                        self.session
                            .record_tool_invocation(&call.name, call.arguments.clone())?,
                    );
                    created_call_box_id = Some(self.session.journal.create_box(
                        now(),
                        call_name,
                        BoxOwner::Kennedy,
                        call_content,
                    )?);
                    let external_event_id = self.session.pending_external_event_id.clone();
                    if self
                        .session
                        .recover_context_overflow(external_event_id.as_deref(), &[])?
                        == ContextRecovery::Irreducible
                    {
                        ToolOutcome {
                            text: CONTEXT_OVERFLOW_WARNING.into(),
                            store_result: false,
                            ok: false,
                            end_session: false,
                            freeform_write: None,
                            managed_source_snapshot: None,
                            exact_result: false,
                        }
                    } else if call.name == LAUNCH_SESSION_TOOL {
                        let invocation = recorded_invocation
                            .as_ref()
                            .context("LaunchSession invocation was not recorded")?;
                        let result =
                            (|| -> anyhow::Result<(LaunchSessionArguments, LaunchIntent)> {
                                self.session.assert_tool_allowed(LAUNCH_SESSION_TOOL)?;
                                let arguments = decode_launch_session_arguments(&call.arguments)?;
                                let intent =
                                    self.session.prepare_launch_intent(invocation, &arguments)?;
                                Ok((arguments, intent))
                            })();
                        match result {
                            Ok((arguments, intent)) => {
                                (self.checkpoint)(self.session.snapshot()?).await?;
                                match self.session.lower_launch(&intent, &arguments).await {
                                    Ok(launch) => ToolOutcome {
                                        text: launch_success_json(
                                            &launch.session_id,
                                            &launch.command_id,
                                        )?,
                                        store_result: true,
                                        ok: true,
                                        end_session: false,
                                        freeform_write: None,
                                        managed_source_snapshot: None,
                                        exact_result: true,
                                    },
                                    Err(error)
                                        if matches!(
                                            error.kind,
                                            HistoryErrorKind::InvalidInput
                                                | HistoryErrorKind::Conflict
                                        ) =>
                                    {
                                        ToolOutcome {
                                            text: format!(
                                                "LaunchSession failed: {}",
                                                error.message
                                            ),
                                            store_result: true,
                                            ok: false,
                                            end_session: false,
                                            freeform_write: None,
                                            managed_source_snapshot: None,
                                            exact_result: false,
                                        }
                                    }
                                    Err(error) => {
                                        return Err(anyhow::anyhow!(
                                            "LaunchSession remains unresolved ({}): {}",
                                            error.kind.code(),
                                            error.message
                                        ));
                                    }
                                }
                            }
                            Err(error) => ToolOutcome {
                                text: format!("LaunchSession failed: {error}"),
                                store_result: true,
                                ok: false,
                                end_session: false,
                                freeform_write: None,
                                managed_source_snapshot: None,
                                exact_result: false,
                            },
                        }
                    } else {
                        match self.session.execute_tool(&call, operation_id).await {
                            Ok(outcome) => {
                                emitted_response = call.name == "EmitObject" && outcome.ok;
                                outcome
                            }
                            Err(error) => ToolOutcome {
                                text: format!("{} failed: {error}", call.name),
                                store_result: call.name != "LoadNodes",
                                ok: false,
                                end_session: false,
                                freeform_write: None,
                                managed_source_snapshot: None,
                                exact_result: false,
                            },
                        }
                    }
                }
                Err(error) => ToolOutcome {
                    text: error.to_string(),
                    store_result: true,
                    ok: false,
                    end_session: false,
                    freeform_write: None,
                    managed_source_snapshot: None,
                    exact_result: false,
                },
            };
            if let Some(snapshot) = outcome.managed_source_snapshot.take() {
                apply_snapshot(&mut self.session.journal, &now(), snapshot)?;
                outcome.store_result = false;
            }
            if !outcome.exact_result {
                append_slow_tool_duration(&mut outcome.text, tool_started_at.elapsed());
            }
            self.state.exact_tool_result = outcome.exact_result;
            let capture = if let Some(request) = outcome.freeform_write.take() {
                self.state.pending_freeform_write = Some(PendingFreeformWrite {
                    request,
                    call_box_id: created_call_box_id
                        .context("freeform write call box was not created")?,
                });
                Some(json!(true))
            } else {
                None
            };
            if outcome.store_result {
                outcome.text = ensure_tool_result_box(
                    &mut self.session.journal,
                    recorded_invocation.as_ref(),
                    &outcome.text,
                    outcome.ok,
                )?;
            }
            let external_event_id = self.session.pending_external_event_id.clone();
            let recovery = self
                .session
                .recover_context_overflow(external_event_id.as_deref(), &[])?;
            let context_warning_added =
                self.session.transcript[transcript_start..]
                    .iter()
                    .any(|entry| {
                        entry.get("contextOverflowWarning").and_then(Value::as_bool) == Some(true)
                    });
            let mut provider_text = outcome.text.clone();
            if !outcome.exact_result
                && context_warning_added
                && !provider_text.contains(CONTEXT_OVERFLOW_WARNING)
            {
                if !provider_text.is_empty() {
                    provider_text.push_str("\n\n");
                }
                provider_text.push_str(CONTEXT_OVERFLOW_WARNING);
            }
            self.session.record_tool_completion(
                recorded_invocation.as_ref(),
                json!({"ok":outcome.ok,"result":outcome.text}),
            )?;
            let stop = recovery == ContextRecovery::Irreducible
                || (matches!(self.session.mode, AgentMode::Ingress { .. })
                    && self.session.ingress_force_commit_requested())
                || (!matches!(self.session.mode, AgentMode::Ingress { .. })
                    && self.session.journal.state().source_terminated);
            Ok(kcode_agent_runtime::SessionToolOutcome {
                text: provider_text,
                ok: outcome.ok,
                capture,
                stop,
                finish_after_round: outcome.end_session,
                emitted_response,
            })
        })
    }

    fn prepare_provider_resume<'a>(
        &'a mut self,
        mut outcome: kcode_agent_runtime::SessionToolOutcome,
    ) -> kcode_agent_runtime::HostFuture<'a, kcode_agent_runtime::ProviderResume> {
        Box::pin(async move {
            if completes_before_provider_resume(&outcome) {
                (self.checkpoint)(self.session.snapshot()?).await?;
                return Ok(kcode_agent_runtime::ProviderResume::Complete(None));
            }

            let ingress_time = self
                .session
                .ingress_time_remaining()?
                .map(|remaining| (remaining, self.session.previous_ingress_attempt_timed_out));
            let synchronized_after = self
                .state
                .provider_synchronized_after
                .context("provider resume was prepared before its input was recorded")?;
            let prepared = self.session.journal.prepare_provider_resume(
                now(),
                synchronized_after,
                ingress_time,
            )?;
            match apply_prepared_provider_resume(
                &mut self.session.provider_affinity,
                &mut self.session.next_thread_reset_reason,
                prepared,
            ) {
                NativeProviderResumePreparation::Continue { marker_lines } => {
                    self.state.provider_synchronized_after = Some(
                        self.session
                            .journal
                            .state()
                            .events
                            .last()
                            .context("provider resume preparation left no journal event")?
                            .id,
                    );
                    if self.state.exact_tool_result {
                        self.state.exact_tool_result = false;
                    } else {
                        let mut footer_lines = marker_lines;
                        footer_lines.extend(self.session.runtime_budget().footer_lines());
                        outcome.text = provider_tool_result_with_context_footer(
                            &footer_lines.join("\n"),
                            &outcome.text,
                        );
                    }
                    (self.checkpoint)(self.session.snapshot()?).await?;
                    Ok(kcode_agent_runtime::ProviderResume::Continue(outcome))
                }
                NativeProviderResumePreparation::RestartFresh { reason } => {
                    self.state.exact_tool_result = false;
                    self.state.restart_fresh_reason = Some(reason);
                    (self.checkpoint)(self.session.snapshot()?).await?;
                    Ok(kcode_agent_runtime::ProviderResume::RestartFresh)
                }
            }
        })
    }

    fn complete_capture<'a>(
        &'a mut self,
        _capture: Value,
        contents: String,
    ) -> kcode_agent_runtime::HostFuture<'a, kcode_agent_runtime::SessionControl> {
        Box::pin(async move {
            let pending = self
                .state
                .pending_freeform_write
                .take()
                .context("provider completed without a pending freeform write")?;
            let result_metadata = pending.request.clone();
            let outcome = self
                .session
                .complete_freeform_write(pending, contents)
                .await?;
            if outcome.store_result {
                self.session.journal.create_box(
                    now(),
                    "Kennedy tool result",
                    BoxOwner::Controller,
                    BoxContent::text(&outcome.text),
                )?;
            }
            self.session.journal.record(
                now(),
                EventKind::Note {
                    label: "write_file_freeform_result".into(),
                    value: result_metadata.result_record(outcome.ok, &outcome.text),
                },
            )?;
            let external_event_id = self.session.pending_external_event_id.clone();
            let recovery = self
                .session
                .recover_context_overflow(external_event_id.as_deref(), &[])?;
            let snapshot = self.session.snapshot()?;
            (self.checkpoint)(snapshot).await?;
            if recovery == ContextRecovery::Irreducible
                || (matches!(self.session.mode, AgentMode::Ingress { .. })
                    && self.session.ingress_force_commit_requested())
                || (!matches!(self.session.mode, AgentMode::Ingress { .. })
                    && self.session.journal.state().source_terminated)
                || self.state.deadline_after_response
            {
                return Ok(kcode_agent_runtime::SessionControl::Complete(None));
            }
            self.session.journal.create_box(
                now(),
                controller_box_name(&self.session.mode),
                BoxOwner::Controller,
                BoxContent::text(controller_message(
                    &self.session.mode,
                    &self.session.free_time,
                )),
            )?;
            Ok(kcode_agent_runtime::SessionControl::Continue)
        })
    }

    fn complete_round<'a>(
        &'a mut self,
        completion: kcode_agent_runtime::RoundCompletion,
    ) -> kcode_agent_runtime::HostFuture<'a, kcode_agent_runtime::SessionControl> {
        Box::pin(async move {
            let answer = completion.answer.trim().to_owned();
            let mut completion_recovery = ContextRecovery::NotNeeded;
            if !answer.is_empty() {
                let mut content = BoxContent::text(answer.clone());
                if let Some(id) = &self.session.pending_external_event_id {
                    content.metadata["externalEventId"] = json!(id);
                }
                self.session.journal.create_box(
                    now(),
                    "Kennedy message",
                    BoxOwner::Kennedy,
                    content,
                )?;
                let mut transcript = json!({"role":"kennedy","content":answer});
                if let Some(id) = &self.session.pending_external_event_id {
                    transcript["externalEventId"] = json!(id);
                }
                self.session.transcript.push(transcript);
                self.session.synchronize_provider_known_events();
                let external_event_id = self.session.pending_external_event_id.clone();
                completion_recovery = self
                    .session
                    .recover_context_overflow(external_event_id.as_deref(), &[])?;
            }
            let snapshot = self.session.snapshot()?;
            (self.checkpoint)(snapshot).await?;
            if completion_recovery == ContextRecovery::Irreducible
                || (matches!(self.session.mode, AgentMode::Ingress { .. })
                    && self.session.ingress_force_commit_requested())
                || (!matches!(self.session.mode, AgentMode::Ingress { .. })
                    && self.session.journal.state().source_terminated)
            {
                return Ok(kcode_agent_runtime::SessionControl::Complete(None));
            }
            if completion.finish_requested || self.state.deadline_after_response {
                return Ok(kcode_agent_runtime::SessionControl::Complete(
                    (!answer.is_empty()).then_some(answer),
                ));
            }
            if matches!(self.session.mode, AgentMode::Conversation) && !answer.is_empty() {
                return Ok(kcode_agent_runtime::SessionControl::Complete(Some(answer)));
            }
            if matches!(self.session.mode, AgentMode::Conversation) && completion.emitted_response {
                return Ok(kcode_agent_runtime::SessionControl::Complete(None));
            }
            let solo_ingress_response =
                matches!(self.session.mode, AgentMode::Ingress { .. }) && !answer.is_empty();
            anyhow::ensure!(
                completion.used_tool || solo_ingress_response,
                "provider completed without a response or tool call"
            );
            self.session.journal.create_box(
                now(),
                controller_box_name(&self.session.mode),
                BoxOwner::Controller,
                BoxContent::text(controller_message(
                    &self.session.mode,
                    &self.session.free_time,
                )),
            )?;
            Ok(kcode_agent_runtime::SessionControl::Continue)
        })
    }
}

impl kcode_agent_runtime::Host for KennedySubagentHost<'_> {
    fn render_tool_call(&mut self, call: &kcode_agent_runtime::ToolCall) -> anyhow::Result<String> {
        Ok(tool_invocation_content(&call.name, &call.arguments)?.text)
    }

    fn execute_tool<'a>(
        &'a mut self,
        call: kcode_agent_runtime::ToolCall,
        operation_id: Uuid,
        budget: kcode_agent_runtime::ContextBudget,
    ) -> kcode_agent_runtime::HostFuture<'a, kcode_agent_runtime::ToolOutcome> {
        Box::pin(async move {
            let call = ToolCall {
                name: call.name,
                arguments: call.arguments,
            };
            if let Some(reason) = subagent_unavailable_reason(&call.name) {
                return Ok(kcode_agent_runtime::ToolOutcome::failure(reason));
            }
            if budget.estimated_tokens() > budget.max_input_tokens() {
                return Ok(kcode_agent_runtime::ToolOutcome::failure(
                    "The Ktool call was not run because its retained invocation would exceed the subagent context limit.",
                ));
            }
            if !subagent_managed_write_fits(&self.context, &call, &budget) {
                return Ok(kcode_agent_runtime::ToolOutcome::failure(
                    "The managed-source write was not run because its resulting current state would exceed the subagent context limit.",
                ));
            }

            let tool_started_at = std::time::Instant::now();

            if call.name == "LoadNodes" {
                let Some(DecodedTool::LoadNodes(identifiers)) =
                    decode(&call.name, &call.arguments)?
                else {
                    return Ok(kcode_agent_runtime::ToolOutcome::failure(
                        "LoadNodes did not match its tool contract.",
                    ));
                };
                load_durable_batch(
                    self.session.api.kmap(),
                    self.context.kweb_mut(),
                    &identifiers,
                )?;
                let (updates, creates) = self.session.plan.context_projection();
                let changes = self.context.reconcile_kweb(&updates, &creates)?;
                let displayed_state_keys = changes.displayed_state_keys();
                let mut text = if changes.is_empty() {
                    "LoadNodes completed. The subagent Kweb projection was already current.".into()
                } else {
                    changes.display_text()
                };
                append_slow_tool_duration(&mut text, tool_started_at.elapsed());
                return Ok(kcode_agent_runtime::ToolOutcome {
                    text,
                    ok: true,
                    state_updates: changes.updates,
                    displayed_state_keys,
                    capture: None,
                });
            }

            if is_kweb_mutation(&call.name) {
                self.session.assert_tool_allowed(&call.name)?;
                let decoded = decode(&call.name, &call.arguments)?
                    .with_context(|| format!("{} did not match its tool contract", call.name))?;
                let prior_create_count = self.session.plan.create_count();
                let (mut text, referenced_pending) = execute_kweb_mutation(
                    &call.name,
                    decoded,
                    self.context.kweb(),
                    &mut self.session.plan,
                    &mut self.session.journal,
                )?;
                self.context.include_staged_nodes(
                    referenced_pending
                        .into_iter()
                        .chain(self.session.plan.pending_ids_from(prior_create_count)),
                );
                let (updates, creates) = self.session.plan.context_projection();
                let changes = self.context.reconcile_kweb(&updates, &creates)?;
                append_slow_tool_duration(&mut text, tool_started_at.elapsed());
                return Ok(kcode_agent_runtime::ToolOutcome {
                    text,
                    ok: true,
                    state_updates: changes.updates,
                    displayed_state_keys: Vec::new(),
                    capture: None,
                });
            }

            if let Some(request) = decode_freeform_write(&call.name, &call.arguments)? {
                if !self.context.source_is_open(request.kind(), request.name()) {
                    return Ok(kcode_agent_runtime::ToolOutcome::failure(format!(
                        "{} {:?} is not open in this subagent context. Call {} first.",
                        request.kind().label(),
                        request.name(),
                        request.kind().open_tool()
                    )));
                }
                let acknowledgement = request.acknowledgement();
                let id = Uuid::new_v4().to_string();
                self.captures.insert(id.clone(), request);
                return Ok(kcode_agent_runtime::ToolOutcome {
                    text: acknowledgement,
                    ok: true,
                    state_updates: Vec::new(),
                    displayed_state_keys: Vec::new(),
                    capture: Some(Value::String(id)),
                });
            }

            let mut outcome = match self.session.execute_tool(&call, operation_id).await {
                Ok(outcome) => outcome,
                Err(error) => {
                    let mut text = format!("{} failed: {error}", call.name);
                    append_slow_tool_duration(&mut text, tool_started_at.elapsed());
                    return Ok(kcode_agent_runtime::ToolOutcome::failure(text));
                }
            };
            let displays_managed_snapshot = outcome
                .managed_source_snapshot
                .as_ref()
                .is_some_and(|snapshot| result_displays_snapshot(&outcome.text, snapshot));
            append_slow_tool_duration(&mut outcome.text, tool_started_at.elapsed());
            let (state_updates, displayed_state_keys) =
                if let Some(snapshot) = outcome.managed_source_snapshot.take() {
                    let state = self.context.apply_source_snapshot(snapshot);
                    let displayed = displays_managed_snapshot.then_some(state.key);
                    (
                        state.update.into_iter().collect(),
                        displayed.into_iter().collect(),
                    )
                } else {
                    (Vec::new(), Vec::new())
                };
            let capture = outcome.freeform_write.take().map(|request| {
                let id = Uuid::new_v4().to_string();
                self.captures.insert(id.clone(), request);
                Value::String(id)
            });
            Ok(kcode_agent_runtime::ToolOutcome {
                text: outcome.text,
                ok: outcome.ok,
                state_updates,
                displayed_state_keys,
                capture,
            })
        })
    }

    fn complete_capture<'a>(
        &'a mut self,
        capture: Value,
        contents: String,
        budget: kcode_agent_runtime::ContextBudget,
    ) -> kcode_agent_runtime::HostFuture<'a, kcode_agent_runtime::ToolOutcome> {
        Box::pin(async move {
            let id = capture
                .as_str()
                .context("subagent freeform capture token is invalid")?;
            let request = self
                .captures
                .remove(id)
                .context("subagent freeform capture token is unknown")?;
            self.session
                .complete_subagent_freeform_write(&mut self.context, request, contents, &budget)
                .await
        })
    }

    fn record(&mut self, event: kcode_agent_runtime::AuditEvent) -> anyhow::Result<()> {
        kcode_intelligence_chatend::record_subagent_event(&mut self.session.journal, &now(), &event)
    }
}

fn cost_summary(label: &str, estimated_cost_usd_nanos: u64, unpriced_calls: u64) -> String {
    render(RenderRequest::CostSummary {
        label,
        estimated_cost_usd_nanos,
        unpriced_calls,
    })
    .expect("cost-summary rendering is infallible")
}

fn restore_kweb_context(journal: &HistorySession, context: &mut KwebContext) -> anyhow::Result<()> {
    let Some(tool) = journal.state().tools.get(KWEB_TOOL_INSTANCE) else {
        return Ok(());
    };
    let mut nodes = BTreeMap::new();
    for slot in &tool.slots {
        let state = journal
            .state()
            .box_state(slot.box_id)
            .context("Kweb slot references a missing box")?;
        if let Some(node) = state.canonical.content.metadata.get("storedNode") {
            let node = match serde_json::from_value::<KwebNode>(node.clone()) {
                Ok(node) => node,
                Err(_) => node_from_value(node).context("decoding a stored Kweb context node")?,
            };
            nodes.insert(node.id.clone(), node);
        }
    }
    let mut direct = journal
        .state()
        .current_ingress_attempt_events()
        .iter()
        .flat_map(|event| {
            let EventKind::ToolInvoked {
                tool_name,
                arguments,
                ..
            } = &event.kind
            else {
                return Vec::new();
            };
            match tool_name.as_str() {
                "LoadNodes" => arguments
                    .get("identifiers")
                    .and_then(Value::as_array)
                    .into_iter()
                    .flatten()
                    .filter_map(Value::as_str)
                    .map(str::to_owned)
                    .collect(),
                "LoadNode" => arguments
                    .get("identifier")
                    .and_then(Value::as_str)
                    .map(str::to_owned)
                    .into_iter()
                    .collect(),
                _ => Vec::new(),
            }
        })
        .collect::<Vec<_>>();
    if direct.is_empty() {
        direct = context.root_node_ids().to_vec();
    }
    context
        .restore(nodes.into_values(), direct)
        .map_err(anyhow::Error::new)
}

fn session_kind(session_type: &str, mode: &AgentMode) -> SessionKind {
    if matches!(mode, AgentMode::Ingress { .. }) {
        return SessionKind::HistoryIngress;
    }
    match session_type {
        "conversation" => SessionKind::Conversation,
        "telegram" => SessionKind::Telegram,
        "telegram-group" => SessionKind::TelegramGroup,
        "free-time" => SessionKind::SelfTime,
        "wakeup" => SessionKind::Other("wakeup".into()),
        "audio" => SessionKind::AudioIngress,
        other => SessionKind::Other(other.into()),
    }
}

fn tool_instance_for_invocation(name: &str, invocation_id: &str) -> String {
    if name == "LoadNodes" {
        return KWEB_TOOL_INSTANCE.into();
    }
    format!("{name}:{invocation_id}")
}

fn canonical_id(value: &str) -> anyhow::Result<String> {
    value
        .parse::<NodeId>()
        .with_context(|| format!("{value:?} is not a canonical node ID"))?;
    Ok(value.into())
}

fn image_extension(media_type: &str) -> &'static str {
    match media_type
        .split(';')
        .next()
        .unwrap_or(media_type)
        .trim()
        .to_ascii_lowercase()
        .as_str()
    {
        "image/jpeg" => "jpg",
        "image/webp" => "webp",
        _ => "png",
    }
}

fn call_ktool_description(include_launch_session: bool) -> String {
    let mut description = render(RenderRequest::CallKtoolDescription)
        .expect("Ktool-description rendering is infallible");
    if include_launch_session {
        description.push_str(
            "\n\nLaunchSession is available for this genuine user turn. Call it with exactly directive (a nonblank string) and contextNodeIds (an ordered array of distinct fully loaded canonical node IDs). It launches an ordinary browser conversation and returns exactly sessionId and commandId.",
        );
    }
    description
}

fn now() -> String {
    Utc::now().to_rfc3339()
}

fn deadline(value: &Value) -> Option<DateTime<Utc>> {
    value
        .get("deadlineAt")
        .and_then(Value::as_str)
        .and_then(|value| DateTime::parse_from_rfc3339(value).ok())
        .map(|value| value.with_timezone(&Utc))
}

fn remaining_until(deadline: DateTime<Utc>) -> Duration {
    (deadline - Utc::now()).to_std().unwrap_or(Duration::ZERO)
}

fn controller_box_name(mode: &AgentMode) -> &'static str {
    match mode {
        AgentMode::Conversation => "Turn continuation",
        AgentMode::FreeTime => "Self-time continuation",
        AgentMode::Wakeup => "Wakeup continuation",
        AgentMode::Ingress { .. } => "History-ingress continuation",
    }
}

fn controller_message(mode: &AgentMode, free_time: &Value) -> String {
    let mode = match mode {
        AgentMode::Conversation => "conversation",
        AgentMode::FreeTime => "free-time",
        AgentMode::Wakeup => "wakeup",
        AgentMode::Ingress { .. } => "ingress",
    };
    render(RenderRequest::ControllerMessage { mode, free_time })
        .expect("known controller modes render successfully")
}

#[cfg(test)]
mod tests {
    use std::time::{SystemTime, UNIX_EPOCH};

    use kcode_session_history::{Config as HistoryConfig, NewSession, SessionHistory};

    use super::*;

    fn test_history(label: &str) -> (std::path::PathBuf, SessionHistory) {
        let root = std::env::temp_dir().join(format!(
            "kcode-kennedy-sessions-{label}-{}-{}",
            std::process::id(),
            SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        let history = SessionHistory::open(HistoryConfig {
            directory: root.join("sessions"),
            completed_list: root.join("completed.jsonl"),
            provider_cost_compatibility: None,
        })
        .unwrap();
        (root, history)
    }

    fn test_journal(label: &str) -> (std::path::PathBuf, HistorySession) {
        let (root, history) = test_history(label);
        let journal = history
            .create_session(NewSession {
                kind: SessionKind::SelfTime,
                created_at: "2026-08-05T00:00:00Z".into(),
                effective_context_tokens: 10_000,
                channel: Value::Null,
            })
            .unwrap();
        (root, journal)
    }

    fn provider_affinity() -> ProviderAffinityState {
        ProviderAffinityState {
            continuation: kcode_intelligence_router::AgentContinuation {
                thread_id: "thread-1".into(),
                provider_model: "gpt-5.6-sol".into(),
                cumulative_input_tokens: 120,
                cumulative_output_tokens: 30,
                cumulative_cached_input_tokens: 80,
                cumulative_reasoning_output_tokens: 10,
            },
            synchronized_event_id: EventId(42),
            material_fingerprint: "material".into(),
        }
    }

    fn launch_intent(invocation_id: &str, turn: u64) -> LaunchIntent {
        LaunchIntent {
            invocation_id: invocation_id.into(),
            user_turn_id: EventId(turn),
            started_at: "2026-08-15T00:00:00Z".into(),
            parent_session_id: "parent".into(),
            effective_context_tokens: 10_000,
            root_node_ids: vec!["AAAAAAAE".into()],
            reference_root_node_ids: Vec::new(),
            context_node_ids: Vec::new(),
        }
    }

    fn launch_invocation(
        journal: &mut HistorySession,
        invocation_id: &str,
    ) -> RecordedToolInvocation {
        let invocation = RecordedToolInvocation {
            invocation_id: invocation_id.into(),
            tool_instance: tool_instance_for_invocation(LAUNCH_SESSION_TOOL, invocation_id),
            tool_name: LAUNCH_SESSION_TOOL.into(),
        };
        journal
            .record(
                now(),
                EventKind::ToolInvoked {
                    tool_instance: invocation.tool_instance.clone(),
                    tool_name: invocation.tool_name.clone(),
                    arguments: json!({"directive":"go","contextNodeIds":[]}),
                    invocation_id: Some(invocation.invocation_id.clone()),
                },
            )
            .unwrap();
        invocation
    }

    fn invocation_result_box_count(journal: &HistorySession, invocation_id: &str) -> usize {
        journal
            .state()
            .boxes
            .values()
            .filter(|state| {
                state
                    .canonical
                    .content
                    .metadata
                    .get("toolInvocationId")
                    .and_then(Value::as_str)
                    == Some(invocation_id)
            })
            .count()
    }

    #[test]
    fn stepped_wait_types_are_send_and_static() {
        fn assert_send_static<T: Send + 'static>() {}
        assert_send_static::<PendingSessionInference>();
        assert_send_static::<SessionInferenceWake>();
    }

    #[test]
    fn turn_lease_rejects_duplicates_invalidates_and_avoids_aba_release() {
        let mut slot = None;
        let first = TurnLease::acquire(&mut slot).unwrap();
        first.validate(&slot).unwrap();
        assert!(TurnLease::acquire(&mut slot).is_err());

        slot = None;
        assert!(first.validate(&slot).is_err());

        let second = TurnLease::acquire(&mut slot).unwrap();
        assert!(first.validate(&slot).is_err());
        second.validate(&slot).unwrap();
        drop(first);
        second.validate(&slot).unwrap();
        assert!(TurnLease::acquire(&mut slot).is_err());

        drop(second);
        let third = TurnLease::acquire(&mut slot).unwrap();
        third.validate(&slot).unwrap();
    }

    #[test]
    fn subagents_reject_parent_controls_but_allow_delivery_effects() {
        for unavailable in [
            LAUNCH_SESSION_TOOL,
            "RunSubagent",
            "EndSession",
            "DehydrateBoxes",
            "SummarizeBox",
            "HydrateBox",
            "BoxesIntoObjects",
        ] {
            assert!(subagent_unavailable_reason(unavailable).is_some());
        }
        for delegated in [
            "NoteToSelf",
            "EmitObject",
            "SendTelegramDM",
            "SendTelegramGroupMessage",
            "LoadNodes",
            "ExtractDocumentText",
        ] {
            assert_eq!(subagent_unavailable_reason(delegated), None);
        }
    }

    #[test]
    fn only_complete_snapshot_results_claim_to_display_managed_state() {
        let snapshot = SourceSnapshot {
            kind: kcode_dev_tools::ManagedSourceKind::RustLibrary,
            name: "example".into(),
            text: "complete source".into(),
        };

        assert!(result_displays_snapshot("complete source", &snapshot));
        assert!(!result_displays_snapshot(
            "Wrote file src/lib.rs in Rust library example.",
            &snapshot
        ));
    }

    #[test]
    fn provider_affinity_round_trips_through_snapshot_json() {
        let affinity = provider_affinity();
        let restored: ProviderAffinityState =
            serde_json::from_value(serde_json::to_value(&affinity).unwrap()).unwrap();
        assert_eq!(restored, affinity);
    }

    #[test]
    fn rewritten_resume_clears_affinity_and_selects_durable_restart() {
        let mut affinity = Some(provider_affinity());
        let mut next_reason = None;
        let action = apply_prepared_provider_resume(
            &mut affinity,
            &mut next_reason,
            PreparedProviderResume {
                marker_lines: vec!["[due marker]".into()],
                thread_reset_reason: Some("provider_history_rewritten".into()),
            },
        );
        assert_eq!(
            action,
            NativeProviderResumePreparation::RestartFresh {
                reason: "provider_history_rewritten".into()
            }
        );
        assert!(affinity.is_none());
        assert_eq!(next_reason.as_deref(), Some("provider_history_rewritten"));
    }

    #[test]
    fn append_only_resume_keeps_markers_affinity_and_synchronization_path() {
        let original = provider_affinity();
        let mut affinity = Some(original.clone());
        let mut next_reason = None;
        let action = apply_prepared_provider_resume(
            &mut affinity,
            &mut next_reason,
            PreparedProviderResume {
                marker_lines: vec!["[one]".into(), "[two]".into()],
                thread_reset_reason: None,
            },
        );
        assert_eq!(
            action,
            NativeProviderResumePreparation::Continue {
                marker_lines: vec!["[one]".into(), "[two]".into()]
            }
        );
        assert_eq!(affinity, Some(original));
        assert!(next_reason.is_none());
    }

    #[test]
    fn restart_next_round_is_one_complete_projection_without_continuation() {
        let (root, mut journal) = test_journal("restart-projection");
        journal
            .create_box(
                now(),
                "Durable tool result",
                BoxOwner::Controller,
                BoxContent::text("durable-result-once"),
            )
            .unwrap();
        let prepared = journal
            .prepare_provider_projection(now(), &[], "material", None)
            .unwrap();
        assert_eq!(prepared.provider_input, prepared.projection.render());
        assert_eq!(
            prepared
                .provider_input
                .matches("durable-result-once")
                .count(),
            1
        );
        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn state_version_four_never_restores_native_affinity() {
        let affinity = provider_affinity();
        let version_four = json!({"stateVersion":4,"providerAffinity":affinity.clone()});
        let version_five = json!({"stateVersion":5,"providerAffinity":affinity.clone()});
        assert!(
            restore_provider_affinity(Some(&version_four), false)
                .unwrap()
                .is_none()
        );
        assert_eq!(
            restore_provider_affinity(Some(&version_five), false).unwrap(),
            Some(affinity)
        );
    }

    #[test]
    fn restart_failures_leave_affinity_cleared_and_sync_unadvanced() {
        let synchronized_after = Some(EventId(7));
        let mut affinity = Some(provider_affinity());
        let mut next_reason = None;
        let _ = apply_prepared_provider_resume(
            &mut affinity,
            &mut next_reason,
            PreparedProviderResume {
                marker_lines: Vec::new(),
                thread_reset_reason: Some("rewrite".into()),
            },
        );
        let receipt_failure: anyhow::Result<()> = Err(anyhow::anyhow!("receipt failed"));
        let fresh_failure: anyhow::Result<()> = Err(anyhow::anyhow!("fresh start failed"));
        assert!(receipt_failure.is_err() && fresh_failure.is_err());
        assert!(affinity.is_none());
        assert_eq!(next_reason.as_deref(), Some("rewrite"));
        assert_eq!(synchronized_after, Some(EventId(7)));
    }

    #[test]
    fn ingress_deadline_starts_at_2700_seconds_rounds_down_and_expires_safely() {
        let now = Instant::now();
        let mut deadline = None;
        assert_eq!(
            ingress_time_remaining_at(&mut deadline, now).unwrap(),
            2_700
        );

        let mut near_deadline = Some(now + Duration::from_millis(1_500));
        assert_eq!(
            ingress_time_remaining_at(&mut near_deadline, now).unwrap(),
            1
        );
        let expired =
            ingress_time_remaining_at(&mut near_deadline, now + Duration::from_millis(1_500))
                .unwrap_err();
        assert!(is_ingress_time_expired(&expired));
    }

    #[test]
    fn successful_end_session_completes_before_another_provider_resume() {
        let mut outcome = kcode_agent_runtime::SessionToolOutcome::success("Session ending.");
        outcome.finish_after_round = true;
        assert!(completes_before_provider_resume(&outcome));

        outcome.ok = false;
        assert!(!completes_before_provider_resume(&outcome));

        outcome.stop = true;
        assert!(completes_before_provider_resume(&outcome));
    }

    #[test]
    fn child_kweb_mutation_changes_leaf_plan_without_creating_parent_boxes() {
        let (root, mut journal) = test_journal("subagent-kweb");
        let node_id = "AAAAAAAE".to_owned();
        let mut context = KwebContext::new(vec![node_id.clone()]).unwrap();
        context
            .apply_load(
                KwebNode {
                    id: node_id.clone(),
                    short_name: "Old".into(),
                    short_description: "Old summary".into(),
                    long_description: "Old details".into(),
                    owner: "self".into(),
                    fixed_connections: Vec::new(),
                    recent_connections: Vec::new(),
                    objects: Vec::new(),
                    last_modified_by: "test".into(),
                    last_modified_at: None,
                },
                Vec::new(),
            )
            .unwrap();
        let mut plan = KwebPlan::default();

        let (result, _) = execute_kweb_mutation(
            "UpdateNode",
            DecodedTool::UpdateNode {
                id: node_id.clone(),
                owner: "self".into(),
                short_name: "New".into(),
                short_description: "New summary".into(),
                long_description: "New details".into(),
            },
            &context,
            &mut plan,
            &mut journal,
        )
        .unwrap();

        assert_eq!(result, format!("Staged the update to node {node_id}."));
        assert_eq!(
            plan.checkpoint_value().unwrap()["updates"][&node_id]["longDescription"],
            "New details"
        );
        assert!(journal.state().boxes.is_empty());
        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn connect_nodes_reports_distinct_final_counts_in_first_input_order() {
        let (root, mut journal) = test_journal("connect-node-counts");
        let ids = ["AAAAAAAE", "AAAAAAAI", "AAAAAAAM"];
        let mut context =
            KwebContext::new(ids.iter().map(|id| (*id).to_owned()).collect::<Vec<_>>()).unwrap();
        for id in ids {
            context
                .apply_load(
                    KwebNode {
                        id: id.into(),
                        short_name: id.into(),
                        short_description: id.into(),
                        long_description: id.into(),
                        owner: "self".into(),
                        fixed_connections: Vec::new(),
                        recent_connections: Vec::new(),
                        objects: Vec::new(),
                        last_modified_by: "test".into(),
                        last_modified_at: None,
                    },
                    Vec::new(),
                )
                .unwrap();
        }
        let mut plan = KwebPlan::default();
        let input = vec![
            "AAAAAAAI".into(),
            "AAAAAAAE".into(),
            "AAAAAAAI".into(),
            "AAAAAAAM".into(),
        ];

        let (result, _) = execute_kweb_mutation(
            "ConnectNodes",
            DecodedTool::ConnectNodes(input),
            &context,
            &mut plan,
            &mut journal,
        )
        .unwrap();

        assert_eq!(
            result,
            "Staged connections among nodes AAAAAAAI, AAAAAAAE, AAAAAAAI, AAAAAAAM.\nPost-call recent connection counts: AAAAAAAI: 2, AAAAAAAE: 3, AAAAAAAM: 3."
        );
        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn launch_arguments_are_strict_unbounded_and_byte_exact() {
        let directive = "  exact directive \n";
        let arguments = decode_launch_session_arguments(&json!({
            "directive":directive,
            "contextNodeIds":[],
        }))
        .unwrap();
        assert_eq!(arguments.directive, directive);
        assert!(
            decode_launch_session_arguments(&json!({
                "directive":"x",
                "contextNodeIds":[],
                "extra":true,
            }))
            .is_err()
        );
        assert!(
            decode_launch_session_arguments(&json!({
                "directive":"   ",
                "contextNodeIds":[],
            }))
            .is_err()
        );
    }

    #[test]
    fn launch_context_ids_reject_duplicates_pending_markers_and_malformed_values() {
        assert!(
            validate_canonical_distinct_ids(
                &["AAAAAAAE".into(), "AAAAAAAE".into()],
                "context node"
            )
            .is_err()
        );
        for invalid in ["pending:1", "[box updated]", "not-an-id"] {
            assert!(validate_canonical_distinct_ids(&[invalid.into()], "context node").is_err());
        }
        assert!(validate_canonical_distinct_ids(&[], "context node").is_ok());
    }

    #[test]
    fn launch_description_is_conditional() {
        assert!(!call_ktool_description(false).contains("LaunchSession is available"));
        assert!(call_ktool_description(true).contains("LaunchSession is available"));
    }

    #[test]
    fn launch_success_has_only_two_identity_fields() {
        let value: Value =
            serde_json::from_str(&launch_success_json("session", "command").unwrap()).unwrap();
        assert_eq!(value, json!({"sessionId":"session","commandId":"command"}));
    }

    #[test]
    fn ten_intents_are_counted_per_turn_and_replay_is_not_new() {
        let current = EventId(10);
        let intents = (0..10)
            .map(|index| launch_intent(&format!("intent-{index}"), current.0))
            .collect::<Vec<_>>();
        assert_eq!(
            intents
                .iter()
                .filter(|intent| intent.user_turn_id == current)
                .count(),
            MAX_LAUNCH_INTENTS_PER_USER_TURN
        );
        assert_eq!(
            intents
                .iter()
                .filter(|intent| intent.invocation_id == "intent-0")
                .count(),
            1
        );
        let next = EventId(11);
        assert_eq!(
            intents
                .iter()
                .filter(|intent| intent.user_turn_id == next)
                .count(),
            0
        );
    }

    #[test]
    fn pruning_retains_current_and_every_unfinished_prior_intent() {
        let intents = vec![
            launch_intent("current-complete", 2),
            launch_intent("prior-complete", 1),
            launch_intent("prior-unfinished", 1),
        ];
        let completed =
            BTreeSet::from(["current-complete".to_owned(), "prior-complete".to_owned()]);
        let retained = pruned_launch_intents(&intents, Some(EventId(2)), &completed);
        assert_eq!(retained.len(), 2);
        assert_eq!(retained[0].invocation_id, "current-complete");
        assert_eq!(retained[1].invocation_id, "prior-unfinished");
        let cleared = pruned_launch_intents(&retained, None, &completed);
        assert_eq!(cleared.len(), 1);
        assert_eq!(cleared[0].invocation_id, "prior-unfinished");
    }

    #[test]
    fn journal_ahead_authority_selects_user_box_not_later_event() {
        let (root, mut journal) = test_journal("launch-user-event");
        let user_box = journal
            .create_box(
                now(),
                "User message",
                BoxOwner::User,
                BoxContent::text("hello"),
            )
            .unwrap();
        journal
            .record(
                now(),
                EventKind::Note {
                    label: "later".into(),
                    value: Value::Null,
                },
            )
            .unwrap();
        let selected = unique_user_box_event(&journal, &journal.state().events).unwrap();
        assert_eq!(selected, Some(EventId(user_box.0)));
        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn journal_ahead_synthetic_bootstrap_consumes_marker_then_later_user_authorizes() {
        let (root, mut journal) = test_journal("launch-bootstrap-recovery");
        let checkpoint_event_count = journal.state().events.len();
        let synthetic_box = journal
            .create_box(
                now(),
                "Synthetic launch directive",
                BoxOwner::User,
                BoxContent::text("bootstrap"),
            )
            .unwrap();
        let recovered =
            unique_user_box_event(&journal, &journal.state().events[checkpoint_event_count..])
                .unwrap();
        assert_eq!(recovered, Some(EventId(synthetic_box.0)));

        let provenance = json!({
            "kind":"synthetic-launch-bootstrap",
            "denyLaunchSession":true,
        });
        let mut orchestration = json!({"launchBootstrapPending":true});
        let mut launch_bootstrap_pending = true;
        let mut launch_user_turn_id = None;
        reconcile_recovered_launch_authority(
            true,
            recovered,
            &provenance,
            &mut orchestration,
            &mut launch_bootstrap_pending,
            &mut launch_user_turn_id,
        );
        assert!(launch_user_turn_id.is_none());
        assert!(!launch_bootstrap_pending);
        assert_eq!(orchestration["launchBootstrapPending"], false);

        let later_start = journal.state().events.len();
        let genuine_box = journal
            .create_box(
                now(),
                "User message",
                BoxOwner::User,
                BoxContent::text("real turn"),
            )
            .unwrap();
        let genuine =
            unique_user_box_event(&journal, &journal.state().events[later_start..]).unwrap();
        let authority =
            user_turn_launch_authority(genuine, &mut orchestration, &mut launch_bootstrap_pending);
        assert_eq!(authority, Some(EventId(genuine_box.0)));
        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn provenance_round_trips_without_granting_authority() {
        let provenance = json!({
            "kind":"synthetic-launch-bootstrap",
            "denyLaunchSession":true,
        });
        let encoded = serde_json::to_value(&provenance).unwrap();
        assert_eq!(encoded, provenance);
        assert_eq!(
            json!({"launchBootstrapPending":false})["launchBootstrapPending"],
            false
        );
    }

    #[test]
    fn post_intent_missing_selected_node_does_not_block_child_or_consume_command() {
        let (root, history) = test_history("launch-post-intent-node-loss");
        let child_id = Uuid::new_v4().to_string();
        let intent = LaunchIntent {
            invocation_id: child_id.clone(),
            user_turn_id: EventId(7),
            started_at: "2026-08-15T00:00:00Z".into(),
            parent_session_id: "parent".into(),
            effective_context_tokens: 10_000,
            root_node_ids: vec!["AAAAAAAE".into()],
            reference_root_node_ids: Vec::new(),
            context_node_ids: vec!["AAAAAAAI".into()],
        };
        let request = Session::launch_request(&intent, "exact directive");
        assert_eq!(request.state["launchContextNodeIds"], json!(["AAAAAAAI"]));

        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let launch = runtime.block_on(history.launch_session(request)).unwrap();
        assert_eq!(launch.session_id, child_id);
        let record = runtime.block_on(history.get(&child_id)).unwrap();
        assert_eq!(record.state["launchContextNodeIds"], json!(["AAAAAAAI"]));
        let commands = runtime.block_on(history.command_heads()).unwrap();
        let command = commands
            .iter()
            .find(|command| command.conversation_id == child_id)
            .unwrap();
        assert_eq!(command.id, launch.command_id);
        assert_eq!(command.status, "pending");
        assert!(!command.cancel_requested);
        let still_pending = runtime.block_on(history.command_heads()).unwrap();
        assert_eq!(
            still_pending
                .iter()
                .filter(|command| command.conversation_id == child_id)
                .count(),
            1
        );
        drop(runtime);
        drop(history);
        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn launch_recovery_reuses_one_result_box_and_records_one_completion() {
        for (label, result, expected_ok, expected_text) in [
            (
                "success",
                Ok(kcode_session_history::SessionLaunch {
                    session_id: "session".into(),
                    command_id: "command".into(),
                }),
                true,
                "{\"sessionId\":\"session\",\"commandId\":\"command\"}",
            ),
            (
                "terminal",
                Err(kcode_session_history::Error {
                    kind: HistoryErrorKind::Conflict,
                    message: "stable conflict".into(),
                }),
                false,
                "LaunchSession failed: stable conflict",
            ),
        ] {
            let (root, mut journal) = test_journal(&format!("launch-result-{label}"));
            let invocation_id = Uuid::new_v4().to_string();
            let invocation = launch_invocation(&mut journal, &invocation_id);
            ensure_tool_result_box(&mut journal, Some(&invocation), expected_text, expected_ok)
                .unwrap();
            complete_launch_reconciliation(&mut journal, &invocation, result).unwrap();
            complete_launch_reconciliation(
                &mut journal,
                &invocation,
                Ok(kcode_session_history::SessionLaunch {
                    session_id: "ignored-after-completion".into(),
                    command_id: "ignored-after-completion".into(),
                }),
            )
            .unwrap();

            assert_eq!(invocation_result_box_count(&journal, &invocation_id), 1);
            let completions = journal
                .state()
                .events
                .iter()
                .filter_map(|event| {
                    let EventKind::ToolCompleted {
                        invocation_id: Some(id),
                        outcome,
                        ..
                    } = &event.kind
                    else {
                        return None;
                    };
                    (id == &invocation_id).then_some(outcome.clone())
                })
                .collect::<Vec<_>>();
            assert_eq!(
                completions,
                vec![json!({"ok":expected_ok,"result":expected_text})]
            );
            std::fs::remove_dir_all(root).unwrap();
        }
    }

    #[test]
    fn unresolved_launch_storage_creates_neither_result_box_nor_completion() {
        let (root, mut journal) = test_journal("launch-storage-unresolved");
        let invocation_id = Uuid::new_v4().to_string();
        let invocation = launch_invocation(&mut journal, &invocation_id);
        let error = complete_launch_reconciliation(
            &mut journal,
            &invocation,
            Err(kcode_session_history::Error {
                kind: HistoryErrorKind::Storage,
                message: "storage unavailable".into(),
            }),
        )
        .unwrap_err();
        assert!(error.to_string().contains("remains unresolved"));
        assert_eq!(invocation_result_box_count(&journal, &invocation_id), 0);
        assert!(!completed_invocation_ids(&journal).contains(&invocation_id));
        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn unfinished_launch_intent_is_distinct_from_pre_intent_invocation() {
        let (root, mut journal) = test_journal("launch-completion-helper");
        journal
            .record(
                now(),
                EventKind::ToolInvoked {
                    tool_instance: "LaunchSession:id".into(),
                    tool_name: LAUNCH_SESSION_TOOL.into(),
                    arguments: json!({"directive":"go","contextNodeIds":[]}),
                    invocation_id: Some("id".into()),
                },
            )
            .unwrap();
        assert!(completed_invocation_ids(&journal).is_empty());
        assert!(invocation_arguments(&journal, "id").is_ok());
        journal.repair_unfinished_tools(now()).unwrap();
        assert!(completed_invocation_ids(&journal).contains("id"));
        std::fs::remove_dir_all(root).unwrap();
    }
}