#![recursion_limit = "256"]

use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use std::sync::{LazyLock, Mutex, RwLock};
use std::time::{Duration, Instant};

use clap::{Args, Subcommand};
use kura_core::tool_capabilities::public_tool_capabilities;
use reqwest::Method;
use serde_json::{Map, Value, json};
use tokio::io::{self, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
use uuid::Uuid;
mod analyze_tool;
mod tool_dedupe;
mod tool_response;
mod tool_schema;
pub mod tool_schema_hashes;
mod util;

#[cfg(test)]
use analyze_tool::{
    ANALYZE_AGENT_DEFAULT_WAIT_TIMEOUT_MS, ANALYZE_WAIT_TIMEOUT_MS_MAX, ANALYZE_WAIT_TIMEOUT_MS_MIN,
};
use analyze_tool::{
    ANALYZE_ENDPOINT, analyze_input_schema, analyze_tool_description, build_analyze_body,
    prepare_agent_analyze_body,
};
#[cfg(test)]
use tool_dedupe::{
    TOOL_CALL_DEDUPE_CACHE, ToolCallDedupeEntry, is_tool_call_dedupe_eligible, tool_call_dedupe_key,
};
use tool_dedupe::{
    TOOL_CALL_DEDUPE_WINDOW_MS, get_tool_call_dedupe_entry, stable_dedupe_args_signature,
    store_tool_call_dedupe_entry,
};
use tool_response::{
    build_tool_call_response, finalize_tool_envelope_for_budget, flatten_public_tool_response,
    public_payload_contract_overflow, public_tool_forbids_payload_truncation,
};
use tool_schema::sanitize_tool_input_schema;
#[cfg(test)]
use tool_schema::{collect_unsupported_schema_keywords, tool_schema_compatibility_summary};
use util::{client, resolve_token};

const MCP_PROTOCOL_VERSION: &str = "2025-06-18";
const MCP_SERVER_NAME: &str = "kura-mcp";
const READ_QUERY_V4_PATH: &str = "/v4/agent/read-query";
const READ_QUERY_REQUEST_SCHEMA_VERSION: &str = "read_query_request.v1";
const RECORD_ACTIVITY_V4_PATH: &str = "/v4/agent/record-activity";
const RECORD_RECOVERY_V4_PATH: &str = "/v4/agent/record-recovery";
const REPORT_HEALTH_V4_PATH: &str = "/v4/agent/report-health";
const RECORD_NUTRITION_V4_PATH: &str = "/v4/agent/record-nutrition";
const RECORD_MEASUREMENT_V4_PATH: &str = "/v4/agent/record-measurement";
const RECORD_BENCHMARK_V4_PATH: &str = "/v4/agent/record-benchmark";
const PLAN_ACTIVITY_V4_PATH: &str = "/v4/agent/plan-activity";
const UPDATE_PROFILE_V4_PATH: &str = "/v4/agent/update-profile";
const UPDATE_PLAN_V4_PATH: &str = "/v4/agent/update-plan";
const AGENT_SKILLPACK_BRIEF_PATH: &str = "/v1/agent/skillpacks/brief";
const KURA_CLIENT_NAME_HEADER: &str = "x-kura-client-name";
const KURA_CLIENT_VERSION_HEADER: &str = "x-kura-client-version";
const KURA_CLIENT_INSTALL_CHANNEL_HEADER: &str = "x-kura-client-install-channel";
const KURA_CLIENT_NOTICE_ACK_HEADER: &str = "x-kura-client-notice-ack";
const KURA_NOTICE_ACK_MAX_IDS: usize = 16;
const TOOL_ENVELOPE_MAX_BYTES: usize = 28_000;
const COMPACT_ENDPOINT_PREVIEW_MAX_ITEMS: usize = 120;
const CONTEXT_SESSION_TTL_SECS: u64 = 3600;
const RETRIEVAL_FSM_WINDOW_SECS: u64 = 90;
const RETRIEVAL_FSM_MAX_RELOADS_PER_WINDOW: u32 = 12;
const RETRIEVAL_FSM_MAX_REPEAT_SIGNATURE_STREAK: u32 = 3;
const RETRIEVAL_FSM_WINDOW_SECS_MIN: u64 = 10;
const RETRIEVAL_FSM_WINDOW_SECS_MAX: u64 = 3600;
const RETRIEVAL_FSM_MAX_RELOADS_PER_WINDOW_MIN: u32 = 1;
const RETRIEVAL_FSM_MAX_RELOADS_PER_WINDOW_MAX: u32 = 200;
const RETRIEVAL_FSM_MAX_REPEAT_SIGNATURE_STREAK_MIN: u32 = 1;
const RETRIEVAL_FSM_MAX_REPEAT_SIGNATURE_STREAK_MAX: u32 = 20;
const RETRIEVAL_FSM_WINDOW_SECS_ENV: &str = "KURA_MCP_RETRIEVAL_FSM_WINDOW_SECS";
const RETRIEVAL_FSM_MAX_RELOADS_ENV: &str = "KURA_MCP_RETRIEVAL_FSM_MAX_RELOADS";
const RETRIEVAL_FSM_MAX_REPEAT_SIGNATURE_STREAK_ENV: &str =
    "KURA_MCP_RETRIEVAL_FSM_MAX_REPEAT_SIGNATURE_STREAK";
const RETRIEVAL_OBSERVABILITY_SCHEMA_VERSION: &str = "mcp_retrieval_observability.v1";
const ROLLOUT_GUARD_SCHEMA_VERSION: &str = "mcp_rollout_guard.v1";
const IMPORT_DEVICE_TOOLS_ENABLED_ENV: &str = "KURA_MCP_ENABLE_IMPORT_PROVIDER_TOOLS";
const DETERMINISM_ROLLOUT_MODE_ENV: &str = "KURA_MCP_DETERMINISM_ROLLOUT_MODE";
const DETERMINISM_CANARY_PERCENT_ENV: &str = "KURA_MCP_DETERMINISM_CANARY_PERCENT";
const DETERMINISM_CANARY_ALLOWLIST_ENV: &str = "KURA_MCP_DETERMINISM_CANARY_ALLOWLIST";
const DETERMINISM_CANARY_USER_ALLOWLIST_ENV: &str = "KURA_MCP_DETERMINISM_CANARY_USER_ALLOWLIST";
const DETERMINISM_CANARY_EMAIL_ALLOWLIST_ENV: &str = "KURA_MCP_DETERMINISM_CANARY_EMAIL_ALLOWLIST";
const DETERMINISM_ABORT_ON_BREACH_ENV: &str = "KURA_MCP_DETERMINISM_ABORT_ON_BREACH";
const DETERMINISM_ABORT_OVERFLOW_RATE_MAX_ENV: &str = "KURA_MCP_ROLLOUT_ABORT_OVERFLOW_RATE_MAX";
const DETERMINISM_ABORT_CRITICAL_MISSING_RATE_MAX_ENV: &str =
    "KURA_MCP_ROLLOUT_ABORT_CRITICAL_MISSING_RATE_MAX";
const DETERMINISM_ABORT_BLOCKED_RATE_MAX_ENV: &str = "KURA_MCP_ROLLOUT_ABORT_BLOCKED_RATE_MAX";
const DETERMINISM_ABORT_SPECULATIVE_RATE_MAX_ENV: &str =
    "KURA_MCP_ROLLOUT_ABORT_SPECULATIVE_RATE_MAX";
const DETERMINISM_MIN_CONTEXT_CALLS_ENV: &str = "KURA_MCP_ROLLOUT_MIN_CONTEXT_CALLS";
const DETERMINISM_MIN_TOTAL_CALLS_ENV: &str = "KURA_MCP_ROLLOUT_MIN_TOTAL_CALLS";
const EMIT_STARTUP_DIAGNOSTICS_ENV: &str = "KURA_MCP_EMIT_STARTUP_DIAGNOSTICS";
const DETERMINISM_DEFAULT_CANARY_PERCENT: u8 = 10;
const DETERMINISM_DEFAULT_ABORT_OVERFLOW_RATE_MAX: f64 = 0.45;
const DETERMINISM_DEFAULT_ABORT_CRITICAL_MISSING_RATE_MAX: f64 = 0.15;
const DETERMINISM_DEFAULT_ABORT_BLOCKED_RATE_MAX: f64 = 0.35;
const DETERMINISM_DEFAULT_ABORT_SPECULATIVE_RATE_MAX: f64 = 0.02;
const DETERMINISM_DEFAULT_MIN_CONTEXT_CALLS: u64 = 3;
const DETERMINISM_DEFAULT_MIN_TOTAL_CALLS: u64 = 10;
#[cfg(test)]
const UNKNOWN_FIELD_HIGH_IMPACT_API_ERROR: &str = "unknown_field_high_impact_opt_in_required";
#[cfg(test)]
const UNKNOWN_FIELD_HIGH_IMPACT_NEXT_ACTION: &str =
    "retry_with_unknown_field_opt_in_or_canonical_fields";
#[cfg(test)]
const UNKNOWN_FIELD_HIGH_IMPACT_FALLBACK_ACTION: &str = "route_to_observation_logged_if_uncertain";
#[cfg(test)]
const CANONICAL_LOGGING_DIRECT_WRITE_API_ERROR: &str =
    "canonical_logging_write_requires_family_write";
#[cfg(test)]
const CANONICAL_LOGGING_DIRECT_WRITE_NEXT_ACTION: &str = "retry_via_family_write";
#[cfg(test)]
const CANONICAL_LOGGING_UNKNOWN_FIELD_API_ERROR: &str = "canonical_logging_unknown_fields_blocked";
#[cfg(test)]
const CANONICAL_LOGGING_UNKNOWN_FIELD_NEXT_ACTION: &str = "rewrite_with_canonical_fields";
#[cfg(test)]
const SET_TYPE_EXPLICIT_MENTION_API_ERROR: &str = "set_type_missing_explicit_mention";
#[cfg(test)]
const SET_TYPE_EXPLICIT_MENTION_NEXT_ACTION: &str = "retry_with_set_type";
#[cfg(test)]
const HIGH_IMPACT_CONFIRMATION_REPLAY_CACHE_TTL_SECS: u64 = 45 * 60;
#[cfg(test)]
const CAPABILITY_PROFILE_REFRESH_TTL_SECS_ENV: &str = "KURA_MCP_CAPABILITY_PROFILE_TTL_SECS";
#[cfg(test)]
const CAPABILITY_PROFILE_REFRESH_TTL_SECS_DEFAULT: i64 = 300;
const EVENTS_LIST_AUTO_PAGINATE_MAX_PAGES_DEFAULT: u64 = 8;
const EVENTS_LIST_AUTO_PAGINATE_MAX_PAGES_MAX: u64 = 50;
const EVENTS_LIST_AUTO_PAGINATE_MAX_TOTAL_DEFAULT: u64 = 1000;
const EVENTS_LIST_AUTO_PAGINATE_MAX_TOTAL_MAX: u64 = 20_000;
/// Tracks which sessions have loaded agent context. Shared across HTTP requests
/// (where each request creates a new McpServer) and stdio (single long-lived server).
/// Keyed by session_id, value is last-seen timestamp for TTL cleanup.
static CONTEXT_SESSIONS: LazyLock<Mutex<HashMap<String, Instant>>> =
    LazyLock::new(|| Mutex::new(HashMap::new()));
static PENDING_NOTICE_ACK_IDS_BY_SESSION: LazyLock<Mutex<HashMap<String, PendingNoticeAckState>>> =
    LazyLock::new(|| Mutex::new(HashMap::new()));
static RETRIEVAL_CONTROL_STATE: LazyLock<Mutex<HashMap<String, RetrievalControlState>>> =
    LazyLock::new(|| Mutex::new(HashMap::new()));
static RETRIEVAL_FSM_POLICY: LazyLock<RetrievalFsmPolicy> =
    LazyLock::new(load_retrieval_fsm_policy_from_env);
static DETERMINISM_ROLLOUT_POLICY: LazyLock<DeterminismRolloutPolicy> =
    LazyLock::new(load_determinism_rollout_policy_from_env);
static SESSION_IDENTITY_CACHE: LazyLock<Mutex<HashMap<String, SessionIdentityCacheEntry>>> =
    LazyLock::new(|| Mutex::new(HashMap::new()));
#[cfg(test)]
static HIGH_IMPACT_CONFIRMATION_REPLAY_CACHE: LazyLock<
    Mutex<HashMap<String, HighImpactConfirmationReplayEntry>>,
> = LazyLock::new(|| Mutex::new(HashMap::new()));
static IMPORT_DEVICE_TOOLS_ENABLED: LazyLock<bool> =
    LazyLock::new(load_import_device_tools_enabled_from_env);

#[derive(Clone, Debug)]
struct RetrievalFsmPolicy {
    window_secs: u64,
    max_reloads_per_window: u32,
    max_repeat_signature_streak: u32,
    configured_via_env: bool,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum DeterminismRolloutMode {
    Off,
    Shadow,
    Canary,
    Full,
}

impl DeterminismRolloutMode {
    fn as_str(self) -> &'static str {
        match self {
            Self::Off => "off",
            Self::Shadow => "shadow",
            Self::Canary => "canary",
            Self::Full => "full",
        }
    }
}

#[derive(Clone, Debug)]
struct RolloutAbortThresholds {
    overflow_rate_max: f64,
    critical_missing_rate_max: f64,
    blocked_rate_max: f64,
    speculative_answer_rate_max: f64,
}

#[derive(Clone, Debug)]
struct DeterminismRolloutPolicy {
    mode: DeterminismRolloutMode,
    canary_percent: u8,
    canary_allowlist: HashSet<String>,
    canary_user_allowlist: HashSet<String>,
    canary_email_allowlist: HashSet<String>,
    abort_on_breach: bool,
    min_context_calls: u64,
    min_total_tool_calls: u64,
    thresholds: RolloutAbortThresholds,
    configured_via_env: bool,
}

#[derive(Clone, Debug)]
struct SessionIdentityCacheEntry {
    last_seen: Instant,
    user_id: Option<String>,
    email: Option<String>,
}

#[cfg(test)]
#[derive(Clone, Debug)]
struct HighImpactConfirmationReplayEntry {
    last_seen: Instant,
    path: String,
    body: Value,
}

impl RetrievalFsmPolicy {
    fn defaults() -> Self {
        Self {
            window_secs: RETRIEVAL_FSM_WINDOW_SECS,
            max_reloads_per_window: RETRIEVAL_FSM_MAX_RELOADS_PER_WINDOW,
            max_repeat_signature_streak: RETRIEVAL_FSM_MAX_REPEAT_SIGNATURE_STREAK,
            configured_via_env: false,
        }
    }
}

#[derive(Clone, Debug)]
struct PendingNoticeAckState {
    last_seen: Instant,
    notice_ids: BTreeSet<String>,
}

#[derive(Clone, Debug)]
struct RetrievalControlState {
    last_seen: Instant,
    window_started_at: Instant,
    reload_count_in_window: u32,
    current_reload_depth: u32,
    max_reload_depth: u32,
    total_reload_depth: u64,
    reload_depth_samples: u64,
    last_retrieval_signature: Option<String>,
    repeated_signature_streak: u32,
    stop_reason: Option<String>,
    total_tool_calls: u64,
    context_loaded_calls: u64,
    context_calls: u64,
    context_overflow_count: u64,
    context_critical_missing_count: u64,
    speculative_answer_sample_count: u64,
    speculative_answer_evidence_count: u64,
    analyze_calls: u64,
    read_query_before_first_analyze_count: u64,
    session_identity_user_id: Option<String>,
    session_identity_email: Option<String>,
    projection_page_calls: u64,
    abort_reasons: BTreeMap<String, u64>,
    pending_analyze_retry: Option<PendingAnalyzeRetryState>,
}

impl RetrievalControlState {
    fn new(now: Instant) -> Self {
        Self {
            last_seen: now,
            window_started_at: now,
            reload_count_in_window: 0,
            current_reload_depth: 0,
            max_reload_depth: 0,
            total_reload_depth: 0,
            reload_depth_samples: 0,
            last_retrieval_signature: None,
            repeated_signature_streak: 0,
            stop_reason: None,
            total_tool_calls: 0,
            context_loaded_calls: 0,
            context_calls: 0,
            context_overflow_count: 0,
            context_critical_missing_count: 0,
            speculative_answer_sample_count: 0,
            speculative_answer_evidence_count: 0,
            analyze_calls: 0,
            read_query_before_first_analyze_count: 0,
            session_identity_user_id: None,
            session_identity_email: None,
            projection_page_calls: 0,
            abort_reasons: BTreeMap::new(),
            pending_analyze_retry: None,
        }
    }
}

#[derive(Clone, Debug)]
struct PendingAnalyzeRetryState {
    next_request: Value,
    reason: Option<String>,
}

#[derive(Clone, Debug)]
struct RetrievalGuardBlock {
    reason_code: &'static str,
    message: &'static str,
    docs_hint: &'static str,
}

fn mark_context_loaded(session_id: &str) {
    let mut map = CONTEXT_SESSIONS.lock().unwrap_or_else(|e| e.into_inner());
    map.insert(session_id.to_string(), Instant::now());
}

fn is_context_loaded(session_id: &str) -> bool {
    let mut map = CONTEXT_SESSIONS.lock().unwrap_or_else(|e| e.into_inner());
    // TTL cleanup on access — remove expired entries
    let cutoff = Instant::now() - std::time::Duration::from_secs(CONTEXT_SESSION_TTL_SECS);
    map.retain(|_, seen| *seen > cutoff);
    map.contains_key(session_id)
}

fn parse_env_bool_flag(raw: Option<String>, default: bool) -> bool {
    match raw {
        Some(value) => matches!(
            value.trim().to_ascii_lowercase().as_str(),
            "1" | "true" | "yes" | "on"
        ),
        None => default,
    }
}

fn emit_startup_diagnostics_enabled() -> bool {
    parse_env_bool_flag(std::env::var(EMIT_STARTUP_DIAGNOSTICS_ENV).ok(), false)
}

fn load_import_device_tools_enabled_from_env() -> bool {
    parse_env_bool_flag(std::env::var(IMPORT_DEVICE_TOOLS_ENABLED_ENV).ok(), false)
}

fn import_device_tools_enabled() -> bool {
    *IMPORT_DEVICE_TOOLS_ENABLED
}

fn parse_env_u64_with_bounds(raw: Option<String>, min: u64, max: u64, default: u64) -> (u64, bool) {
    match raw.and_then(|value| value.parse::<u64>().ok()) {
        Some(parsed) => (parsed.clamp(min, max), true),
        None => (default, false),
    }
}

fn parse_env_u32_with_bounds(raw: Option<String>, min: u32, max: u32, default: u32) -> (u32, bool) {
    match raw.and_then(|value| value.parse::<u32>().ok()) {
        Some(parsed) => (parsed.clamp(min, max), true),
        None => (default, false),
    }
}

fn parse_env_f64_with_bounds(raw: Option<String>, min: f64, max: f64, default: f64) -> (f64, bool) {
    match raw {
        Some(value) => match value.trim().parse::<f64>() {
            Ok(parsed) => (parsed.clamp(min, max), true),
            Err(_) => (default, true),
        },
        None => (default, false),
    }
}

fn parse_allowlist_entries(raw: Option<String>) -> (HashSet<String>, bool) {
    match raw {
        Some(value) => {
            let configured = !value.trim().is_empty();
            let entries = value
                .split([',', ';'])
                .flat_map(|chunk| chunk.split_whitespace())
                .map(str::trim)
                .filter(|entry| !entry.is_empty())
                .map(ToString::to_string)
                .collect::<HashSet<_>>();
            (entries, configured)
        }
        None => (HashSet::new(), false),
    }
}

fn normalize_user_id_for_allowlist(raw: &str) -> Option<String> {
    let trimmed = raw.trim();
    if trimmed.is_empty() {
        return None;
    }
    Some(trimmed.to_ascii_lowercase())
}

fn normalize_email_for_allowlist(raw: &str) -> Option<String> {
    let trimmed = raw.trim();
    if trimmed.is_empty() {
        return None;
    }
    Some(trimmed.to_ascii_lowercase())
}

fn parse_user_allowlist(raw: Option<String>) -> (HashSet<String>, bool) {
    let (entries, configured) = parse_allowlist_entries(raw);
    let normalized = entries
        .into_iter()
        .filter_map(|entry| normalize_user_id_for_allowlist(&entry))
        .collect::<HashSet<_>>();
    (normalized, configured)
}

fn parse_email_allowlist(raw: Option<String>) -> (HashSet<String>, bool) {
    let (entries, configured) = parse_allowlist_entries(raw);
    let normalized = entries
        .into_iter()
        .filter_map(|entry| normalize_email_for_allowlist(&entry))
        .collect::<HashSet<_>>();
    (normalized, configured)
}

fn parse_determinism_rollout_mode(raw: Option<String>) -> (DeterminismRolloutMode, bool) {
    match raw {
        Some(value) => {
            let normalized = value.trim().to_ascii_lowercase();
            if normalized.is_empty() {
                return (DeterminismRolloutMode::Full, true);
            }
            let mode = match normalized.as_str() {
                "off" => DeterminismRolloutMode::Off,
                "shadow" => DeterminismRolloutMode::Shadow,
                "canary" => DeterminismRolloutMode::Canary,
                "full" => DeterminismRolloutMode::Full,
                _ => DeterminismRolloutMode::Full,
            };
            (mode, true)
        }
        None => (DeterminismRolloutMode::Full, false),
    }
}

fn parse_determinism_canary_allowlist(raw: Option<String>) -> (HashSet<String>, bool) {
    parse_allowlist_entries(raw)
}

fn parse_determinism_rollout_policy_from_raw(
    mode_raw: Option<String>,
    canary_percent_raw: Option<String>,
    canary_allowlist_raw: Option<String>,
    canary_user_allowlist_raw: Option<String>,
    canary_email_allowlist_raw: Option<String>,
    abort_on_breach_raw: Option<String>,
    overflow_rate_max_raw: Option<String>,
    critical_missing_rate_max_raw: Option<String>,
    blocked_rate_max_raw: Option<String>,
    speculative_rate_max_raw: Option<String>,
    min_context_calls_raw: Option<String>,
    min_total_tool_calls_raw: Option<String>,
) -> DeterminismRolloutPolicy {
    let (mode, mode_set) = parse_determinism_rollout_mode(mode_raw);
    let (canary_percent_raw, canary_percent_set) = parse_env_u64_with_bounds(
        canary_percent_raw,
        0,
        100,
        u64::from(DETERMINISM_DEFAULT_CANARY_PERCENT),
    );
    let canary_percent =
        u8::try_from(canary_percent_raw).unwrap_or(DETERMINISM_DEFAULT_CANARY_PERCENT);
    let (canary_allowlist, canary_allowlist_set) =
        parse_determinism_canary_allowlist(canary_allowlist_raw);
    let (canary_user_allowlist, canary_user_allowlist_set) =
        parse_user_allowlist(canary_user_allowlist_raw);
    let (canary_email_allowlist, canary_email_allowlist_set) =
        parse_email_allowlist(canary_email_allowlist_raw);
    let abort_on_breach = parse_env_bool_flag(abort_on_breach_raw.clone(), false);
    let abort_on_breach_set = abort_on_breach_raw.is_some();
    let (overflow_rate_max, overflow_rate_set) = parse_env_f64_with_bounds(
        overflow_rate_max_raw,
        0.0,
        1.0,
        DETERMINISM_DEFAULT_ABORT_OVERFLOW_RATE_MAX,
    );
    let (critical_missing_rate_max, critical_missing_set) = parse_env_f64_with_bounds(
        critical_missing_rate_max_raw,
        0.0,
        1.0,
        DETERMINISM_DEFAULT_ABORT_CRITICAL_MISSING_RATE_MAX,
    );
    let (blocked_rate_max, blocked_rate_set) = parse_env_f64_with_bounds(
        blocked_rate_max_raw,
        0.0,
        1.0,
        DETERMINISM_DEFAULT_ABORT_BLOCKED_RATE_MAX,
    );
    let (speculative_answer_rate_max, speculative_rate_set) = parse_env_f64_with_bounds(
        speculative_rate_max_raw,
        0.0,
        1.0,
        DETERMINISM_DEFAULT_ABORT_SPECULATIVE_RATE_MAX,
    );
    let (min_context_calls, min_context_calls_set) = parse_env_u64_with_bounds(
        min_context_calls_raw,
        1,
        10_000,
        DETERMINISM_DEFAULT_MIN_CONTEXT_CALLS,
    );
    let (min_total_tool_calls, min_total_calls_set) = parse_env_u64_with_bounds(
        min_total_tool_calls_raw,
        1,
        10_000,
        DETERMINISM_DEFAULT_MIN_TOTAL_CALLS,
    );

    DeterminismRolloutPolicy {
        mode,
        canary_percent,
        canary_allowlist,
        canary_user_allowlist,
        canary_email_allowlist,
        abort_on_breach,
        min_context_calls,
        min_total_tool_calls,
        thresholds: RolloutAbortThresholds {
            overflow_rate_max,
            critical_missing_rate_max,
            blocked_rate_max,
            speculative_answer_rate_max,
        },
        configured_via_env: mode_set
            || canary_percent_set
            || canary_allowlist_set
            || canary_user_allowlist_set
            || canary_email_allowlist_set
            || abort_on_breach_set
            || overflow_rate_set
            || critical_missing_set
            || blocked_rate_set
            || speculative_rate_set
            || min_context_calls_set
            || min_total_calls_set,
    }
}

fn load_determinism_rollout_policy_from_env() -> DeterminismRolloutPolicy {
    parse_determinism_rollout_policy_from_raw(
        std::env::var(DETERMINISM_ROLLOUT_MODE_ENV).ok(),
        std::env::var(DETERMINISM_CANARY_PERCENT_ENV).ok(),
        std::env::var(DETERMINISM_CANARY_ALLOWLIST_ENV).ok(),
        std::env::var(DETERMINISM_CANARY_USER_ALLOWLIST_ENV).ok(),
        std::env::var(DETERMINISM_CANARY_EMAIL_ALLOWLIST_ENV).ok(),
        std::env::var(DETERMINISM_ABORT_ON_BREACH_ENV).ok(),
        std::env::var(DETERMINISM_ABORT_OVERFLOW_RATE_MAX_ENV).ok(),
        std::env::var(DETERMINISM_ABORT_CRITICAL_MISSING_RATE_MAX_ENV).ok(),
        std::env::var(DETERMINISM_ABORT_BLOCKED_RATE_MAX_ENV).ok(),
        std::env::var(DETERMINISM_ABORT_SPECULATIVE_RATE_MAX_ENV).ok(),
        std::env::var(DETERMINISM_MIN_CONTEXT_CALLS_ENV).ok(),
        std::env::var(DETERMINISM_MIN_TOTAL_CALLS_ENV).ok(),
    )
}

fn determinism_rollout_policy() -> &'static DeterminismRolloutPolicy {
    &DETERMINISM_ROLLOUT_POLICY
}

fn parse_retrieval_fsm_policy_from_raw(
    window_raw: Option<String>,
    max_reloads_raw: Option<String>,
    max_repeat_raw: Option<String>,
) -> RetrievalFsmPolicy {
    let defaults = RetrievalFsmPolicy::defaults();
    let (window_secs, window_set) = parse_env_u64_with_bounds(
        window_raw,
        RETRIEVAL_FSM_WINDOW_SECS_MIN,
        RETRIEVAL_FSM_WINDOW_SECS_MAX,
        defaults.window_secs,
    );
    let (max_reloads_per_window, max_reloads_set) = parse_env_u32_with_bounds(
        max_reloads_raw,
        RETRIEVAL_FSM_MAX_RELOADS_PER_WINDOW_MIN,
        RETRIEVAL_FSM_MAX_RELOADS_PER_WINDOW_MAX,
        defaults.max_reloads_per_window,
    );
    let (max_repeat_signature_streak, max_repeat_set) = parse_env_u32_with_bounds(
        max_repeat_raw,
        RETRIEVAL_FSM_MAX_REPEAT_SIGNATURE_STREAK_MIN,
        RETRIEVAL_FSM_MAX_REPEAT_SIGNATURE_STREAK_MAX,
        defaults.max_repeat_signature_streak,
    );
    RetrievalFsmPolicy {
        window_secs,
        max_reloads_per_window,
        max_repeat_signature_streak,
        configured_via_env: window_set || max_reloads_set || max_repeat_set,
    }
}

fn load_retrieval_fsm_policy_from_env() -> RetrievalFsmPolicy {
    parse_retrieval_fsm_policy_from_raw(
        std::env::var(RETRIEVAL_FSM_WINDOW_SECS_ENV).ok(),
        std::env::var(RETRIEVAL_FSM_MAX_RELOADS_ENV).ok(),
        std::env::var(RETRIEVAL_FSM_MAX_REPEAT_SIGNATURE_STREAK_ENV).ok(),
    )
}

fn retrieval_fsm_policy() -> &'static RetrievalFsmPolicy {
    &RETRIEVAL_FSM_POLICY
}

fn with_retrieval_state_mut<R>(
    session_id: &str,
    f: impl FnOnce(&mut RetrievalControlState, Instant) -> R,
) -> R {
    let now = Instant::now();
    let mut map = RETRIEVAL_CONTROL_STATE
        .lock()
        .unwrap_or_else(|e| e.into_inner());
    let cutoff = now - Duration::from_secs(CONTEXT_SESSION_TTL_SECS);
    map.retain(|_, state| state.last_seen > cutoff);
    let state = map
        .entry(session_id.to_string())
        .or_insert_with(|| RetrievalControlState::new(now));
    state.last_seen = now;
    f(state, now)
}

fn reset_retrieval_window(state: &mut RetrievalControlState, now: Instant) {
    state.window_started_at = now;
    state.reload_count_in_window = 0;
    state.current_reload_depth = 0;
    state.last_retrieval_signature = None;
    state.repeated_signature_streak = 0;
    state.stop_reason = None;
}

fn saturating_ratio(numerator: u64, denominator: u64) -> f64 {
    if denominator == 0 {
        return 0.0;
    }
    let raw = numerator as f64 / denominator as f64;
    (raw * 10_000.0).round() / 10_000.0
}

fn is_retrieval_guarded_tool(name: &str) -> bool {
    matches!(
        name,
        "read_query"
            | "kura_projection_list"
            | "kura_projection_get"
            | "kura_system_manifest"
            | "kura_system_section_get"
    )
}

fn retrieval_signature(name: &str, args: &Map<String, Value>) -> String {
    format!("{name}|{}", stable_dedupe_args_signature(args))
}

fn observe_tool_call_start(session_id: &str, name: &str, context_loaded: bool) {
    with_retrieval_state_mut(session_id, |state, now| {
        state.total_tool_calls = state.total_tool_calls.saturating_add(1);
        if name == "analyze" {
            state.analyze_calls = state.analyze_calls.saturating_add(1);
        } else if name == "read_query" && state.analyze_calls == 0 {
            state.read_query_before_first_analyze_count = state
                .read_query_before_first_analyze_count
                .saturating_add(1);
        }
        if context_loaded {
            state.context_loaded_calls = state.context_loaded_calls.saturating_add(1);
        }
        if !is_retrieval_guarded_tool(name) {
            reset_retrieval_window(state, now);
        }
    });
}

fn record_abort_reason(state: &mut RetrievalControlState, reason: &str) {
    let entry = state.abort_reasons.entry(reason.to_string()).or_insert(0);
    *entry = entry.saturating_add(1);
}

fn record_abort_reason_for_session(session_id: &str, reason: &str) {
    with_retrieval_state_mut(session_id, |state, _| {
        record_abort_reason(state, reason);
    });
}

fn value_at_paths_as_str<'a>(source: &'a Value, paths: &[&str]) -> Option<&'a str> {
    paths
        .iter()
        .find_map(|path| source.pointer(path))
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
}

fn value_at_paths_as_bool(source: &Value, paths: &[&str]) -> Option<bool> {
    paths
        .iter()
        .find_map(|path| source.pointer(path))
        .and_then(Value::as_bool)
}

fn classify_speculative_answer_evidence(name: &str, envelope: &Value) -> Option<bool> {
    if matches!(
        name,
        "record_activity"
            | "record_recovery"
            | "report_health"
            | "record_nutrition"
            | "record_benchmark"
            | "record_measurement"
            | "update_profile"
            | "plan_activity"
            | "update_plan"
    ) {
        let allows_definitive_saved_phrase = value_at_paths_as_bool(
            envelope,
            &[
                "/data/response/body/save_state/allows_definitive_saved_phrase",
                "/data/contract/save_state/allows_definitive_saved_phrase",
            ],
        );
        if let Some(value) = allows_definitive_saved_phrase {
            return Some(!value);
        }
    }

    if !matches!(
        name,
        "record_activity"
            | "record_recovery"
            | "report_health"
            | "record_nutrition"
            | "record_benchmark"
            | "record_measurement"
            | "update_profile"
            | "plan_activity"
            | "update_plan"
    ) {
        return None;
    }

    // Preferred grounded signal from post-response verifier.
    let certainty_state = value_at_paths_as_str(
        envelope,
        &[
            "/data/response/body/post_task_reflection/certainty_state",
            "/data/contract/post_task_reflection/certainty_state",
        ],
    );
    if let Some(state) = certainty_state {
        return Some(!state.eq_ignore_ascii_case("confirmed"));
    }

    // Backward-compatible fallback for older responses without post_task_reflection.
    let verification_status = value_at_paths_as_str(
        envelope,
        &[
            "/data/contract/verification/status",
            "/data/response/body/verification/status",
        ],
    );
    let claim_status = value_at_paths_as_str(
        envelope,
        &[
            "/data/contract/claim_guard/claim_status",
            "/data/response/body/claim_guard/claim_status",
        ],
    );
    let allow_saved_claim = value_at_paths_as_bool(
        envelope,
        &[
            "/data/contract/claim_guard/allow_saved_claim",
            "/data/response/body/claim_guard/allow_saved_claim",
        ],
    );

    if verification_status.is_none() && claim_status.is_none() && allow_saved_claim.is_none() {
        return None;
    }

    let verification_unverified =
        verification_status.is_some_and(|status| !status.eq_ignore_ascii_case("verified"));
    let claim_not_saved =
        claim_status.is_some_and(|status| !status.eq_ignore_ascii_case("saved_verified"));
    let claim_blocked = allow_saved_claim.is_some_and(|allowed| !allowed);

    Some(verification_unverified || claim_not_saved || claim_blocked)
}

fn record_session_identity_for_rollout(
    session_id: &str,
    user_id: Option<String>,
    email: Option<String>,
) {
    with_retrieval_state_mut(session_id, |state, _| {
        state.session_identity_user_id = user_id;
        state.session_identity_email = email;
    });
}

fn get_cached_session_identity(session_id: &str) -> Option<(Option<String>, Option<String>)> {
    let mut cache = SESSION_IDENTITY_CACHE
        .lock()
        .unwrap_or_else(|e| e.into_inner());
    let cutoff = Instant::now() - Duration::from_secs(CONTEXT_SESSION_TTL_SECS);
    cache.retain(|_, entry| entry.last_seen > cutoff);
    cache.get(session_id).map(|entry| {
        (
            entry.user_id.clone(),
            entry.email.clone().map(|value| value.to_ascii_lowercase()),
        )
    })
}

fn cache_session_identity(session_id: &str, user_id: Option<String>, email: Option<String>) {
    let mut cache = SESSION_IDENTITY_CACHE
        .lock()
        .unwrap_or_else(|e| e.into_inner());
    cache.insert(
        session_id.to_string(),
        SessionIdentityCacheEntry {
            last_seen: Instant::now(),
            user_id,
            email: email.map(|value| value.to_ascii_lowercase()),
        },
    );
}

fn rollout_policy_requires_identity(policy: &DeterminismRolloutPolicy) -> bool {
    policy.mode == DeterminismRolloutMode::Canary
        && (!policy.canary_user_allowlist.is_empty() || !policy.canary_email_allowlist.is_empty())
}

fn rollout_bucket_for_session(session_id: &str) -> u8 {
    use std::hash::{Hash, Hasher};
    let mut hasher = std::collections::hash_map::DefaultHasher::new();
    session_id.hash(&mut hasher);
    (hasher.finish() % 100) as u8
}

fn session_in_rollout_canary(
    policy: &DeterminismRolloutPolicy,
    session_id: &str,
    state: &RetrievalControlState,
) -> (bool, &'static str) {
    if policy.canary_allowlist.contains(session_id) {
        return (true, "session_allowlist");
    }
    if state
        .session_identity_user_id
        .as_deref()
        .is_some_and(|value| policy.canary_user_allowlist.contains(value))
    {
        return (true, "user_allowlist");
    }
    if state
        .session_identity_email
        .as_deref()
        .is_some_and(|value| policy.canary_email_allowlist.contains(value))
    {
        return (true, "email_allowlist");
    }
    if rollout_bucket_for_session(session_id) < policy.canary_percent {
        return (true, "bucket");
    }
    (false, "excluded")
}

fn rollout_enforced_for_session(
    policy: &DeterminismRolloutPolicy,
    session_id: &str,
    state: &RetrievalControlState,
) -> bool {
    match policy.mode {
        DeterminismRolloutMode::Off | DeterminismRolloutMode::Shadow => false,
        DeterminismRolloutMode::Canary => session_in_rollout_canary(policy, session_id, state).0,
        DeterminismRolloutMode::Full => true,
    }
}

fn blocked_total_for_state(state: &RetrievalControlState) -> u64 {
    state.abort_reasons.values().copied().sum::<u64>()
}

fn build_rollout_guard_payload_for_policy(
    policy: &DeterminismRolloutPolicy,
    session_id: &str,
    state: &RetrievalControlState,
) -> Value {
    let session_bucket = rollout_bucket_for_session(session_id);
    let (session_in_canary, canary_match_source) =
        session_in_rollout_canary(policy, session_id, state);
    let enforced_for_session = rollout_enforced_for_session(policy, session_id, state);
    let blocked_total = blocked_total_for_state(state);
    let overflow_rate = saturating_ratio(state.context_overflow_count, state.context_calls);
    let critical_missing_rate =
        saturating_ratio(state.context_critical_missing_count, state.context_calls);
    let blocked_rate = saturating_ratio(blocked_total, state.total_tool_calls);
    let speculative_answer_rate = saturating_ratio(
        state.speculative_answer_evidence_count,
        state.speculative_answer_sample_count,
    );

    let context_sample_ready = state.context_calls >= policy.min_context_calls;
    let total_sample_ready = state.total_tool_calls >= policy.min_total_tool_calls;
    let data_sufficient = context_sample_ready && total_sample_ready;

    let mut breached_thresholds = Vec::<Value>::new();
    if context_sample_ready && overflow_rate > policy.thresholds.overflow_rate_max {
        breached_thresholds.push(json!({
            "metric": "overflow_rate",
            "observed": overflow_rate,
            "threshold": policy.thresholds.overflow_rate_max,
            "sample_count": state.context_calls
        }));
    }
    if context_sample_ready && critical_missing_rate > policy.thresholds.critical_missing_rate_max {
        breached_thresholds.push(json!({
            "metric": "critical_missing_rate",
            "observed": critical_missing_rate,
            "threshold": policy.thresholds.critical_missing_rate_max,
            "sample_count": state.context_calls
        }));
    }
    if total_sample_ready && blocked_rate > policy.thresholds.blocked_rate_max {
        breached_thresholds.push(json!({
            "metric": "blocked_rate",
            "observed": blocked_rate,
            "threshold": policy.thresholds.blocked_rate_max,
            "sample_count": state.total_tool_calls
        }));
    }
    if state.speculative_answer_sample_count > 0
        && total_sample_ready
        && speculative_answer_rate > policy.thresholds.speculative_answer_rate_max
    {
        breached_thresholds.push(json!({
            "metric": "speculative_answer_rate",
            "observed": speculative_answer_rate,
            "threshold": policy.thresholds.speculative_answer_rate_max,
            "sample_count": state.speculative_answer_sample_count
        }));
    }

    let status = match policy.mode {
        DeterminismRolloutMode::Off => "disabled",
        _ if !enforced_for_session => "shadow",
        _ if !data_sufficient => "monitoring",
        _ if breached_thresholds.is_empty() => "healthy",
        _ if policy.abort_on_breach => "abort_enforced",
        _ => "abort_recommended",
    };
    let abort_required = status == "abort_enforced";

    json!({
        "schema_version": ROLLOUT_GUARD_SCHEMA_VERSION,
        "mode": policy.mode.as_str(),
        "configured_via_env": policy.configured_via_env,
        "session_bucket": session_bucket,
        "session_in_canary": session_in_canary,
        "canary_match_source": canary_match_source,
        "enforced_for_session": enforced_for_session,
        "identity": {
            "resolved_user_id": state.session_identity_user_id.is_some(),
            "resolved_email": state.session_identity_email.is_some(),
            "allowlist_sizes": {
                "session": policy.canary_allowlist.len(),
                "user": policy.canary_user_allowlist.len(),
                "email": policy.canary_email_allowlist.len()
            }
        },
        "abort_on_breach": policy.abort_on_breach,
        "abort_required": abort_required,
        "sample_requirements": {
            "min_context_calls": policy.min_context_calls,
            "min_total_tool_calls": policy.min_total_tool_calls
        },
        "thresholds": {
            "overflow_rate_max": policy.thresholds.overflow_rate_max,
            "critical_missing_rate_max": policy.thresholds.critical_missing_rate_max,
            "blocked_rate_max": policy.thresholds.blocked_rate_max,
            "speculative_answer_rate_max": policy.thresholds.speculative_answer_rate_max
        },
        "observed": {
            "overflow_rate": overflow_rate,
            "critical_missing_rate": critical_missing_rate,
            "blocked_rate": blocked_rate,
            "speculative_answer_rate": speculative_answer_rate,
            "context_calls": state.context_calls,
            "total_tool_calls": state.total_tool_calls,
            "speculative_answer_sample_count": state.speculative_answer_sample_count
        },
        "data_sufficient": data_sufficient,
        "status": status,
        "rollout_decision": status,
        "breached_thresholds": breached_thresholds,
        "rollback_switch": {
            "env": DETERMINISM_ROLLOUT_MODE_ENV,
            "safe_values": ["shadow", "off"],
            "abort_toggle_env": DETERMINISM_ABORT_ON_BREACH_ENV
        }
    })
}

fn maybe_rollout_abort_guard_block_for_policy(
    policy: &DeterminismRolloutPolicy,
    session_id: &str,
    _tool_name: &str,
    state: &RetrievalControlState,
) -> Option<Value> {
    let guard = build_rollout_guard_payload_for_policy(policy, session_id, state);
    if guard
        .get("status")
        .and_then(Value::as_str)
        .is_some_and(|status| status == "abort_enforced")
    {
        Some(guard)
    } else {
        None
    }
}

fn maybe_rollout_abort_guard_block(session_id: &str, tool_name: &str) -> Option<Value> {
    let policy = determinism_rollout_policy();
    with_retrieval_state_mut(session_id, |state, _| {
        maybe_rollout_abort_guard_block_for_policy(policy, session_id, tool_name, state)
    })
}

fn retrieval_guard_for_reason(reason_code: &str) -> RetrievalGuardBlock {
    match reason_code {
        "max_reloads_exceeded" => RetrievalGuardBlock {
            reason_code: "max_reloads_exceeded",
            message: "Retrieval loop guard blocked more reload attempts in this window.",
            docs_hint: "Switch to a narrower query, use next_cursor progression, or continue with available context before reloading.",
        },
        "repeated_reload_signature" => RetrievalGuardBlock {
            reason_code: "repeated_reload_signature",
            message: "Retrieval loop guard blocked repeated identical reload calls.",
            docs_hint: "Do not retry the same retrieval signature. Advance cursor or change scope before retry.",
        },
        _ => RetrievalGuardBlock {
            reason_code: "loop_guard_reentry_blocked",
            message: "Retrieval loop guard is active for this repeated call signature.",
            docs_hint: "Adjust retrieval arguments before retrying.",
        },
    }
}

fn maybe_block_retrieval_loop(
    session_id: &str,
    name: &str,
    args: &Map<String, Value>,
    policy: &RetrievalFsmPolicy,
) -> Option<RetrievalGuardBlock> {
    if !is_retrieval_guarded_tool(name) {
        return None;
    }

    with_retrieval_state_mut(session_id, |state, now| {
        if now.duration_since(state.window_started_at) > Duration::from_secs(policy.window_secs) {
            reset_retrieval_window(state, now);
        }

        let signature = retrieval_signature(name, args);
        if let Some(active_reason) = state.stop_reason.clone() {
            if state.last_retrieval_signature.as_deref() == Some(signature.as_str()) {
                record_abort_reason(state, &active_reason);
                return Some(retrieval_guard_for_reason(&active_reason));
            }
            state.stop_reason = None;
            state.repeated_signature_streak = 0;
        }

        state.reload_count_in_window = state.reload_count_in_window.saturating_add(1);
        state.current_reload_depth = state.current_reload_depth.saturating_add(1);
        state.max_reload_depth = state.max_reload_depth.max(state.current_reload_depth);
        state.total_reload_depth = state
            .total_reload_depth
            .saturating_add(u64::from(state.current_reload_depth));
        state.reload_depth_samples = state.reload_depth_samples.saturating_add(1);

        if state.last_retrieval_signature.as_deref() == Some(signature.as_str()) {
            state.repeated_signature_streak = state.repeated_signature_streak.saturating_add(1);
        } else {
            state.repeated_signature_streak = 1;
            state.last_retrieval_signature = Some(signature);
        }

        if state.repeated_signature_streak > policy.max_repeat_signature_streak {
            state.stop_reason = Some("repeated_reload_signature".to_string());
            record_abort_reason(state, "repeated_reload_signature");
            return Some(retrieval_guard_for_reason("repeated_reload_signature"));
        }

        if state.reload_count_in_window > policy.max_reloads_per_window {
            state.stop_reason = Some("max_reloads_exceeded".to_string());
            record_abort_reason(state, "max_reloads_exceeded");
            return Some(retrieval_guard_for_reason("max_reloads_exceeded"));
        }

        None
    })
}

fn required_analyze_retry_from_body(body: &Value) -> Option<Value> {
    let next_request = body
        .get("next_request")
        .or_else(|| body.pointer("/suggested_next_action/next_request"))?;
    let action_is_retry = next_request
        .get("action")
        .and_then(Value::as_str)
        .is_some_and(|action| action == "retry_analyze");
    let must_call_before_read_query = next_request
        .get("must_call_before_read_query")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    if action_is_retry && must_call_before_read_query {
        Some(next_request.clone())
    } else {
        None
    }
}

fn required_analyze_retry_from_envelope(envelope: &Value) -> Option<Value> {
    envelope
        .pointer("/data/response/body")
        .and_then(required_analyze_retry_from_body)
}

fn maybe_block_pending_analyze_retry(session_id: &str, name: &str) -> Option<Value> {
    if name != "read_query" {
        return None;
    }
    with_retrieval_state_mut(session_id, |state, _| {
        let pending = state.pending_analyze_retry.clone()?;
        record_abort_reason(state, "analyze_retry_required");
        Some(json!({
            "reason_code": "analyze_retry_required",
            "blocked_tool": "read_query",
            "next_action": "retry_analyze",
            "message": "A previous analyze response returned a required analyze retry. Call that analyze retry before using read_query.",
            "docs_hint": "Call analyze with details.next_request. This guard keeps analysis flows on the same surface instead of silently falling back to raw reads.",
            "must_call_before_read_query": true,
            "reason": pending.reason,
            "next_request": pending.next_request
        }))
    })
}

fn envelope_uses_projection_paging(envelope: &Value) -> bool {
    envelope
        .pointer("/data/request/path")
        .and_then(Value::as_str)
        .map(|path| path.ends_with("/paged"))
        .unwrap_or(false)
}

fn observe_tool_outcome(session_id: &str, name: &str, envelope: &Value) {
    with_retrieval_state_mut(session_id, |state, _| {
        if name == "analyze" {
            state.pending_analyze_retry =
                required_analyze_retry_from_envelope(envelope).map(|next_request| {
                    let reason = next_request
                        .get("reason")
                        .and_then(Value::as_str)
                        .map(str::to_string);
                    PendingAnalyzeRetryState {
                        next_request,
                        reason,
                    }
                });
        }
        if name == "kura_projection_list" && envelope_uses_projection_paging(envelope) {
            state.projection_page_calls = state.projection_page_calls.saturating_add(1);
        }
        if let Some(is_speculative) = classify_speculative_answer_evidence(name, envelope) {
            state.speculative_answer_sample_count =
                state.speculative_answer_sample_count.saturating_add(1);
            if is_speculative {
                state.speculative_answer_evidence_count =
                    state.speculative_answer_evidence_count.saturating_add(1);
            }
        }
    });
}

fn observe_tool_error(session_id: &str, payload: &Value) {
    if let Some(reason) = payload.get("error").and_then(Value::as_str) {
        record_abort_reason_for_session(session_id, reason);
    }
}

fn retrieval_observability_snapshot(session_id: &str, policy: &RetrievalFsmPolicy) -> Value {
    with_retrieval_state_mut(session_id, |state, _| {
        let blocked_total = blocked_total_for_state(state);
        let rollout_guard =
            build_rollout_guard_payload_for_policy(determinism_rollout_policy(), session_id, state);
        json!({
            "schema_version": RETRIEVAL_OBSERVABILITY_SCHEMA_VERSION,
            "fsm": {
                "window_seconds": policy.window_secs,
                "max_reloads_per_window": policy.max_reloads_per_window,
                "max_repeat_signature_streak": policy.max_repeat_signature_streak,
                "configured_via_env": policy.configured_via_env,
                "reload_count_in_window": state.reload_count_in_window,
                "current_reload_depth": state.current_reload_depth,
                "max_reload_depth": state.max_reload_depth,
                "stop_reason": state.stop_reason.clone()
            },
            "metrics": {
                "total_tool_calls": state.total_tool_calls,
                "context_loaded_calls": state.context_loaded_calls,
                "context_hit_rate": saturating_ratio(state.context_loaded_calls, state.total_tool_calls),
                "context_calls": state.context_calls,
                "context_overflow_count": state.context_overflow_count,
                "overflow_rate": saturating_ratio(state.context_overflow_count, state.context_calls),
                "context_critical_missing_count": state.context_critical_missing_count,
                "critical_missing_rate": saturating_ratio(state.context_critical_missing_count, state.context_calls),
                "projection_page_calls": state.projection_page_calls,
                "avg_reload_depth": saturating_ratio(state.total_reload_depth, state.reload_depth_samples),
                "blocked_rate": saturating_ratio(blocked_total, state.total_tool_calls),
                "speculative_answer_sample_count": state.speculative_answer_sample_count,
                "speculative_answer_evidence_count": state.speculative_answer_evidence_count,
                // Backward-compat alias kept for dashboards that still read the old key.
                "speculative_answer_signal_count": state.speculative_answer_evidence_count,
                "speculative_answer_rate": saturating_ratio(
                    state.speculative_answer_evidence_count,
                    state.speculative_answer_sample_count
                ),
                "analyze_calls": state.analyze_calls,
                "read_query_before_first_analyze_count": state.read_query_before_first_analyze_count,
                "analysis_entry_detour_detected": state.read_query_before_first_analyze_count > 0,
                "pending_analyze_retry_required": state.pending_analyze_retry.is_some(),
                "abort_reasons": state.abort_reasons.clone()
            },
            "rollout_guard": rollout_guard
        })
    })
}

fn attach_runtime_observability(
    session_id: &str,
    envelope: &mut Value,
    policy: &RetrievalFsmPolicy,
) {
    if let Some(obj) = envelope.as_object_mut() {
        obj.insert(
            "runtime_observability".to_string(),
            retrieval_observability_snapshot(session_id, policy),
        );
    }
}

#[cfg(test)]
fn clear_retrieval_state(session_id: &str) {
    let mut map = RETRIEVAL_CONTROL_STATE
        .lock()
        .unwrap_or_else(|e| e.into_inner());
    map.remove(session_id);
}

#[derive(Subcommand)]
pub enum McpCommands {
    /// Run a Kura MCP server over stdio
    Serve(McpServeArgs),
}

#[derive(Args, Clone, Debug)]
pub struct McpServeArgs {
    /// Disable auth header injection (useful behind auth proxies)
    #[arg(long)]
    pub no_auth: bool,
    /// Explicit bearer token override (otherwise KURA_API_KEY or OAuth store)
    #[arg(long, env = "KURA_MCP_TOKEN")]
    pub token: Option<String>,
    /// Allow admin API paths (disabled by default)
    #[arg(long, env = "KURA_MCP_ALLOW_ADMIN")]
    pub allow_admin: bool,
}

pub async fn run(api_url: &str, inherited_no_auth: bool, command: McpCommands) -> i32 {
    match command {
        McpCommands::Serve(args) => {
            let mut server = McpServer::new(McpRuntimeConfig {
                api_url: api_url.to_string(),
                no_auth: inherited_no_auth || args.no_auth,
                explicit_token: args.token,
                allow_admin: args.allow_admin,
            });
            match server.serve_stdio().await {
                Ok(()) => 0,
                Err(err) => {
                    let payload = json!({
                        "error": "mcp_server_error",
                        "message": err,
                    });
                    eprintln!("{}", to_pretty_json(&payload));
                    1
                }
            }
        }
    }
}

#[derive(Clone, Debug)]
pub struct HttpMcpRequestConfig {
    pub no_auth: bool,
    pub token: Option<String>,
    pub allow_admin: bool,
}

impl Default for HttpMcpRequestConfig {
    fn default() -> Self {
        Self {
            no_auth: false,
            token: None,
            allow_admin: false,
        }
    }
}

pub async fn handle_http_jsonrpc(
    api_url: &str,
    config: HttpMcpRequestConfig,
    incoming: Value,
) -> Vec<Value> {
    let server = McpServer::new(McpRuntimeConfig {
        api_url: api_url.to_string(),
        no_auth: config.no_auth,
        explicit_token: config.token,
        allow_admin: config.allow_admin,
    });
    server.set_capability_profile(
        match tokio::time::timeout(
            Duration::from_secs(8),
            server.negotiate_capability_profile(),
        )
        .await
        {
            Ok(profile) => profile,
            Err(_) => CapabilityProfile::fallback(
                "capability_negotiation_timeout",
                vec![
                    "Capability negotiation timed out after 8s. Using legacy fallback.".to_string(),
                ],
                None,
            ),
        },
    );
    server.handle_incoming_message(incoming).await
}

#[derive(Clone, Debug)]
struct McpRuntimeConfig {
    api_url: String,
    no_auth: bool,
    explicit_token: Option<String>,
    allow_admin: bool,
}

struct McpServer {
    config: McpRuntimeConfig,
    http: reqwest::Client,
    capability_profile: RwLock<CapabilityProfile>,
    session_id: String,
}

#[derive(Clone, Debug, PartialEq, Eq)]
enum CapabilityMode {
    PreferredContract,
    LegacyFallback,
}

impl CapabilityMode {
    fn as_str(&self) -> &'static str {
        match self {
            CapabilityMode::PreferredContract => "preferred_contract",
            CapabilityMode::LegacyFallback => "legacy_fallback",
        }
    }
}

#[derive(Clone, Debug)]
struct CapabilityProfile {
    mode: CapabilityMode,
    negotiated_at: chrono::DateTime<chrono::Utc>,
    reason: String,
    #[cfg(test)]
    preferred_read_endpoint: String,
    preferred_startup_endpoint: String,
    preferred_section_fetch_endpoint: String,
    #[cfg(test)]
    legacy_read_endpoint: String,
    manifest_snapshot: Option<Value>,
    warnings: Vec<String>,
}

impl CapabilityProfile {
    fn preferred(
        _read: String,
        startup: String,
        section_fetch: String,
        manifest_snapshot: Value,
        warnings: Vec<String>,
    ) -> Self {
        Self {
            mode: CapabilityMode::PreferredContract,
            negotiated_at: chrono::Utc::now(),
            reason: "capabilities_manifest_ok".to_string(),
            #[cfg(test)]
            preferred_read_endpoint: _read,
            preferred_startup_endpoint: startup,
            preferred_section_fetch_endpoint: section_fetch,
            #[cfg(test)]
            legacy_read_endpoint: "/v1/projections".to_string(),
            manifest_snapshot: Some(manifest_snapshot),
            warnings,
        }
    }

    fn fallback(
        reason: impl Into<String>,
        warnings: Vec<String>,
        manifest_snapshot: Option<Value>,
    ) -> Self {
        let reason = reason.into();
        Self {
            mode: CapabilityMode::LegacyFallback,
            negotiated_at: chrono::Utc::now(),
            reason,
            #[cfg(test)]
            preferred_read_endpoint: "/v1/agent/context".to_string(),
            preferred_startup_endpoint: "/v1/agent/context/section-index".to_string(),
            preferred_section_fetch_endpoint: "/v1/agent/context/section-fetch".to_string(),
            #[cfg(test)]
            legacy_read_endpoint: "/v1/projections".to_string(),
            manifest_snapshot,
            warnings,
        }
    }

    #[cfg(test)]
    fn effective_read_endpoint(&self) -> &str {
        if self.mode == CapabilityMode::PreferredContract {
            &self.preferred_read_endpoint
        } else {
            &self.legacy_read_endpoint
        }
    }

    fn to_value(&self) -> Value {
        let mut payload = json!({
            "mode": self.mode.as_str(),
            "reason": self.reason,
            "negotiated_at": self.negotiated_at,
            "preferred_startup_endpoint": self.preferred_startup_endpoint,
            "preferred_section_fetch_endpoint": self.preferred_section_fetch_endpoint,
            "default_logging_tool": "record_activity",
            "public_write_tools": json!(vnext_public_surface_tool_names()),
            "compatibility_turn_tools": json!([]),
            "maintenance_surface_visibility": "explicit_only",
            "public_write_world": "task_first_truth_backed",
        });
        if !self.warnings.is_empty() {
            payload["warnings"] =
                Value::Array(self.warnings.iter().cloned().map(Value::String).collect());
        }
        if let Some(manifest_snapshot) = &self.manifest_snapshot {
            payload["manifest_snapshot"] = sanitize_public_manifest_snapshot(manifest_snapshot);
        }
        payload
    }
}

fn sanitize_public_manifest_snapshot(snapshot: &Value) -> Value {
    json!({
        "min_cli_version": snapshot.get("min_cli_version").cloned().unwrap_or(Value::Null),
        "min_mcp_version": snapshot.get("min_mcp_version").cloned().unwrap_or(Value::Null),
        "protocol_version": snapshot.get("protocol_version").cloned().unwrap_or(Value::Null),
        "preferred_startup_endpoint": snapshot
            .get("preferred_startup_endpoint")
            .cloned()
            .unwrap_or(Value::Null),
        "preferred_section_fetch_endpoint": snapshot
            .get("preferred_section_fetch_endpoint")
            .cloned()
            .unwrap_or(Value::Null),
        "required_verification_contract": snapshot
            .get("required_verification_contract")
            .cloned()
            .unwrap_or(Value::Null),
        "public_write_tools": json!(vnext_public_surface_tool_names()),
        "compatibility_turn_tools": json!([]),
        "public_write_world": "task_first_truth_backed",
    })
}

impl McpServer {
    fn capability_profile_snapshot(&self) -> CapabilityProfile {
        self.capability_profile
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .clone()
    }

    fn set_capability_profile(&self, profile: CapabilityProfile) {
        *self
            .capability_profile
            .write()
            .unwrap_or_else(|poisoned| poisoned.into_inner()) = profile;
    }

    #[cfg(test)]
    fn capability_profile_refresh_ttl() -> chrono::Duration {
        let configured = std::env::var(CAPABILITY_PROFILE_REFRESH_TTL_SECS_ENV)
            .ok()
            .and_then(|raw| raw.trim().parse::<i64>().ok())
            .unwrap_or(CAPABILITY_PROFILE_REFRESH_TTL_SECS_DEFAULT)
            .max(5);
        chrono::Duration::seconds(configured)
    }

    #[cfg(test)]
    fn capability_profile_is_stale(profile: &CapabilityProfile) -> bool {
        chrono::Utc::now().signed_duration_since(profile.negotiated_at)
            >= Self::capability_profile_refresh_ttl()
    }

    fn new(config: McpRuntimeConfig) -> Self {
        // Session ID: credential-derived for HTTP (stable across requests),
        // random UUID for stdio (one server = one session).
        let session_id = match &config.explicit_token {
            Some(token) => format!("sid-{:x}", {
                use std::hash::{Hash, Hasher};
                let mut h = std::collections::hash_map::DefaultHasher::new();
                token.hash(&mut h);
                h.finish()
            }),
            None => format!("stdio-{}", Uuid::now_v7()),
        };
        Self {
            config,
            http: client(),
            capability_profile: RwLock::new(CapabilityProfile::fallback(
                "not_negotiated_yet",
                Vec::new(),
                None,
            )),
            session_id,
        }
    }

    async fn serve_stdio(&mut self) -> Result<(), String> {
        self.set_capability_profile(
            match tokio::time::timeout(Duration::from_secs(8), self.negotiate_capability_profile())
                .await
            {
                Ok(profile) => profile,
                Err(_) => CapabilityProfile::fallback(
                    "capability_negotiation_timeout",
                    vec![
                        "Capability negotiation timed out after 8s. Using legacy fallback."
                            .to_string(),
                    ],
                    None,
                ),
            },
        );
        if emit_startup_diagnostics_enabled() {
            self.emit_capability_status();
        }

        let stdin = io::stdin();
        let mut reader = BufReader::new(stdin);
        let mut stdout = io::stdout();

        loop {
            let incoming = read_framed_json(&mut reader)
                .await
                .map_err(|e| format!("Failed to read MCP message: {e}"))?;
            let Some(incoming) = incoming else {
                break;
            };

            let responses = self.handle_incoming_message(incoming).await;
            for response in responses {
                write_framed_json(&mut stdout, &response)
                    .await
                    .map_err(|e| format!("Failed to write MCP response: {e}"))?;
            }
        }

        Ok(())
    }

    fn emit_capability_status(&self) {
        let capability_profile = self.capability_profile_snapshot();
        let payload = json!({
            "event": "mcp_capability_negotiation",
            "server": MCP_SERVER_NAME,
            "version": env!("CARGO_PKG_VERSION"),
            "profile": capability_profile.to_value(),
        });
        eprintln!("{}", to_pretty_json(&payload));
    }

    async fn negotiate_capability_profile(&self) -> CapabilityProfile {
        let result = self
            .send_api_request(
                Method::GET,
                "/v1/agent/capabilities",
                &[],
                None,
                true,
                false,
            )
            .await;
        capability_profile_from_negotiation(result)
    }

    async fn handle_incoming_message(&self, incoming: Value) -> Vec<Value> {
        let mut responses = Vec::new();

        if let Some(batch) = incoming.as_array() {
            if batch.is_empty() {
                responses.push(error_response(
                    Value::Null,
                    RpcError::invalid_request("Batch request must not be empty"),
                ));
                return responses;
            }
            for item in batch {
                if let Some(response) = self.handle_single_message(item.clone()).await {
                    responses.push(response);
                }
            }
            return responses;
        }

        if let Some(response) = self.handle_single_message(incoming).await {
            responses.push(response);
        }
        responses
    }

    async fn handle_single_message(&self, incoming: Value) -> Option<Value> {
        let Some(obj) = incoming.as_object() else {
            return Some(error_response(
                Value::Null,
                RpcError::invalid_request("Request must be a JSON object"),
            ));
        };

        if obj.get("jsonrpc").and_then(Value::as_str) != Some("2.0") {
            let id = obj.get("id").cloned().unwrap_or(Value::Null);
            return Some(error_response(
                id,
                RpcError::invalid_request("jsonrpc must be '2.0'"),
            ));
        }

        let Some(method) = obj.get("method").and_then(Value::as_str) else {
            // Most likely a client response; server does not issue outbound requests.
            return None;
        };

        let params = obj.get("params").cloned().unwrap_or(Value::Null);
        if let Some(id) = obj.get("id").cloned() {
            let result = self.handle_request(method, params).await;
            Some(match result {
                Ok(payload) => success_response(id, payload),
                Err(err) => error_response(id, err),
            })
        } else {
            self.handle_notification(method, params).await;
            None
        }
    }

    async fn handle_notification(&self, method: &str, _params: Value) {
        if matches!(
            method,
            "notifications/initialized" | "notifications/cancelled"
        ) {
            return;
        }
        // Unknown notifications are intentionally ignored.
    }

    async fn handle_request(&self, method: &str, params: Value) -> Result<Value, RpcError> {
        match method {
            "initialize" => Ok(self.initialize_payload().await),
            "ping" => Ok(json!({})),
            "tools/list" => Ok(self.tools_list_payload()),
            "tools/call" => self.handle_tools_call(params).await,
            "resources/list" => Ok(self.resources_list_payload()),
            "resources/read" => self.handle_resources_read(params).await,
            "prompts/list" => Ok(json!({ "prompts": [] })),
            _ => Err(RpcError::method_not_found(method)),
        }
    }

    async fn initialize_payload(&self) -> Value {
        let skillpack_brief = self.fetch_skillpack_brief().await;
        self.initialize_payload_with_skillpack_brief(skillpack_brief.as_deref())
    }

    fn initialize_payload_with_skillpack_brief(&self, skillpack_brief: Option<&str>) -> Value {
        let capability_profile = self.capability_profile_snapshot();
        let instructions = append_skillpack_brief_to_initialize_instructions(
            format!(
                "This MCP surface is agent-only and intentionally minimal. Use record_activity for executed activity writes, plan_activity for future or not-yet-executed workouts, record_recovery/report_health/record_nutrition/record_measurement/record_benchmark for the matching log tasks, update_profile and update_plan for their matching stateful tasks, read_query for canonical reads, and analyze for bounded deep analysis. For trend, comparison, association, explanation, stagnation, driver, influence, cross-domain, or decision questions, call analyze first with the best public request; do not use read_query or analyze_catalog as routine preflight. Use only the published tools and their schemas; there are no alternate public routes, bootstrap rituals, discovery helpers, or generic raw-write fallbacks in the normal flow. The published schemas are the source of truth: send only canonical fields from the selected tool schema and keep corrections on the same task tool when that schema supports them. For reads, use only the canonical source_id values shown by read_query and continue with the returned cursor when continuation.has_more=true. Treat status=success as a successful save. If a tool result includes user_notice_relay.status=relay_required, append user_notice_relay.append_to_user_reply once in your next user-facing reply after answering the request. Capability mode: {}. Harness profile: task_first_vnext.",
                capability_profile.mode.as_str()
            ),
            skillpack_brief,
        );
        json!({
            "protocolVersion": MCP_PROTOCOL_VERSION,
            "capabilities": {
                "tools": {
                    "listChanged": false
                },
                "resources": {
                    "listChanged": false
                },
                "prompts": {
                    "listChanged": false
                }
            },
            "serverInfo": {
                "name": MCP_SERVER_NAME,
                "version": env!("CARGO_PKG_VERSION")
            },
            "instructions": instructions,
            "capabilityStatus": capability_profile.to_value()
        })
    }

    async fn fetch_skillpack_brief(&self) -> Option<String> {
        let result = self
            .send_api_request(
                Method::GET,
                AGENT_SKILLPACK_BRIEF_PATH,
                &[("surface".to_string(), "mcp".to_string())],
                None,
                true,
                false,
            )
            .await
            .ok()?;
        if !result.is_success() {
            return None;
        }
        result
            .body
            .get("brief")
            .and_then(Value::as_str)
            .map(str::trim)
            .filter(|brief| !brief.is_empty())
            .map(ToString::to_string)
    }

    fn tools_list_payload(&self) -> Value {
        let tools: Vec<Value> = published_tool_definitions()
            .into_iter()
            .map(|tool| {
                json!({
                    "name": tool.name,
                    "description": tool.description,
                    "inputSchema": tool.input_schema,
                    "capabilities": public_tool_capabilities(tool.name),
                })
            })
            .collect();
        json!({ "tools": tools })
    }

    async fn handle_tools_call(&self, params: Value) -> Result<Value, RpcError> {
        let params = params
            .as_object()
            .ok_or_else(|| RpcError::invalid_params("tools/call params must be an object"))?;

        let name = params
            .get("name")
            .and_then(Value::as_str)
            .ok_or_else(|| RpcError::invalid_params("tools/call requires string field 'name'"))?;

        let args = match params.get("arguments") {
            Some(Value::Object(map)) => map.clone(),
            Some(Value::Null) | None => Map::new(),
            Some(_) => {
                return Err(RpcError::invalid_params(
                    "tools/call 'arguments' must be an object",
                ));
            }
        };

        if !is_tool_exposed_on_public_surface(name) {
            return Err(RpcError::invalid_params(format!(
                "Unknown public tool '{name}'. Use one of the tools published by tools/list."
            )));
        }

        let context_loaded = is_context_loaded(&self.session_id);
        observe_tool_call_start(&self.session_id, name, context_loaded);
        let retrieval_policy = retrieval_fsm_policy();

        if rollout_policy_requires_identity(determinism_rollout_policy()) {
            self.resolve_session_identity_for_rollout().await;
        }

        if let Some(rollout_guard) = maybe_rollout_abort_guard_block(&self.session_id, name) {
            record_abort_reason_for_session(&self.session_id, "rollout_abort_guard_blocked");
            let mut envelope = finalize_tool_envelope_for_budget(
                name,
                json!({
                    "status": "error",
                    "phase": "blocked_precondition",
                    "tool": name,
                    "error": {
                        "error": "rollout_abort_guard_blocked",
                        "message": "Deterministic rollout abort criteria were breached for this session. Tool calls are blocked until rollout mode is reduced or guard thresholds are adjusted.",
                        "field": "tool",
                        "docs_hint": "Set KURA_MCP_DETERMINISM_ROLLOUT_MODE=shadow (or off) to rollback safely, then resume once rollout metrics recover.",
                        "details": {
                            "reason_code": "rollout_abort_guard_blocked",
                            "blocked_tool": name,
                            "rollout_guard": rollout_guard
                        }
                    }
                }),
            );
            attach_runtime_observability(&self.session_id, &mut envelope, retrieval_policy);
            return Ok(build_tool_call_response(name, envelope, true));
        }

        if let Some(pending_retry) = maybe_block_pending_analyze_retry(&self.session_id, name) {
            let message = pending_retry
                .get("message")
                .and_then(Value::as_str)
                .unwrap_or("A required analyze retry is pending before read_query.")
                .to_string();
            let docs_hint = pending_retry
                .get("docs_hint")
                .and_then(Value::as_str)
                .unwrap_or("Call analyze with details.next_request before read_query.")
                .to_string();
            let mut envelope = finalize_tool_envelope_for_budget(
                name,
                json!({
                    "status": "error",
                    "phase": "blocked_precondition",
                    "tool": name,
                    "error": {
                        "error": "analyze_retry_required",
                        "message": message,
                        "field": "tool",
                        "docs_hint": docs_hint,
                        "details": pending_retry
                    }
                }),
            );
            attach_runtime_observability(&self.session_id, &mut envelope, retrieval_policy);
            return Ok(build_tool_call_response(name, envelope, true));
        }

        if let Some(guard_block) =
            maybe_block_retrieval_loop(&self.session_id, name, &args, retrieval_policy)
        {
            let mut envelope = finalize_tool_envelope_for_budget(
                name,
                json!({
                    "status": "error",
                    "phase": "blocked_precondition",
                    "tool": name,
                    "error": {
                        "error": "retrieval_loop_guard_blocked",
                        "message": guard_block.message,
                        "field": "arguments",
                        "docs_hint": guard_block.docs_hint,
                        "details": {
                            "reason_code": guard_block.reason_code,
                            "reload_window_seconds": retrieval_policy.window_secs,
                            "max_reloads_per_window": retrieval_policy.max_reloads_per_window,
                            "max_repeat_signature_streak": retrieval_policy.max_repeat_signature_streak,
                            "next_action": "Use narrower retrieval scope or advance next_cursor before retrying."
                        }
                    }
                }),
            );
            attach_runtime_observability(&self.session_id, &mut envelope, retrieval_policy);
            return Ok(build_tool_call_response(name, envelope, true));
        }

        if let Some((mut envelope, age_ms)) =
            get_tool_call_dedupe_entry(&self.session_id, name, &args)
        {
            envelope["dedupe"] = json!({
                "applied": true,
                "reason": "burst_retry_coalesced",
                "window_ms": TOOL_CALL_DEDUPE_WINDOW_MS,
                "age_ms": age_ms
            });
            attach_runtime_observability(&self.session_id, &mut envelope, retrieval_policy);
            return Ok(build_tool_call_response(name, envelope, false));
        }

        let result = self.execute_tool(name, &args).await;
        Ok(match result {
            Ok(payload) => {
                let status = tool_completion_status(&payload);
                let mut envelope = finalize_tool_envelope_for_budget(
                    name,
                    json!({
                        "status": status,
                        "phase": "final",
                        "tool": name,
                        "data": payload
                    }),
                );
                mark_context_loaded(&self.session_id);
                observe_tool_outcome(&self.session_id, name, &envelope);
                store_tool_call_dedupe_entry(&self.session_id, name, &args, &envelope);
                attach_runtime_observability(&self.session_id, &mut envelope, retrieval_policy);
                build_tool_call_response(name, envelope, false)
            }
            Err(err) => {
                let payload = err.to_value();
                observe_tool_error(&self.session_id, &payload);
                let mut envelope = finalize_tool_envelope_for_budget(
                    name,
                    json!({
                        "status": "error",
                        "phase": "final",
                        "tool": name,
                        "error": payload
                    }),
                );
                attach_runtime_observability(&self.session_id, &mut envelope, retrieval_policy);
                build_tool_call_response(name, envelope, true)
            }
        })
    }

    async fn execute_tool(
        &self,
        tool_name: &str,
        args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        match tool_name {
            "kura_api_request" => self.tool_api_request(args).await,
            "read_query" => self.tool_read_query_vnext(args).await,
            "record_activity" => self.tool_record_activity_vnext(args).await,
            "update_profile" => self.tool_update_profile_vnext(args).await,
            "record_recovery" => self.tool_record_recovery_vnext(args).await,
            "report_health" => self.tool_report_health_vnext(args).await,
            "record_nutrition" => self.tool_record_nutrition_vnext(args).await,
            "record_benchmark" => self.tool_record_benchmark_vnext(args).await,
            "record_measurement" => self.tool_record_measurement_vnext(args).await,
            "plan_activity" => self.tool_plan_activity_vnext(args).await,
            "update_plan" => self.tool_update_plan_vnext(args).await,
            "kura_events_list" => self.tool_events_list(args).await,
            "kura_projection_get" => self.tool_projection_get(args).await,
            "kura_projection_list" => self.tool_projection_list(args).await,
            "kura_system_manifest" => self.tool_system_manifest(args).await,
            "kura_system_section_get" => self.tool_system_section_get(args).await,
            "kura_agent_answer_admissibility" => self.tool_agent_answer_admissibility(args).await,
            "kura_semantic_resolve" => self.tool_semantic_resolve(args).await,
            "kura_exercise_vocabulary_discover" => {
                self.tool_exercise_vocabulary_discover(args).await
            }
            "kura_access_request" => self.tool_access_request(args).await,
            "kura_account_api_keys_list" => self.tool_account_api_keys_list(args).await,
            "kura_account_api_keys_create" => self.tool_account_api_keys_create(args).await,
            "kura_account_api_keys_revoke" => self.tool_account_api_keys_revoke(args).await,
            "kura_import_job_create" => {
                if import_device_tools_enabled() {
                    self.tool_import_job_create(args).await
                } else {
                    Err(ToolError::new(
                        "capability_not_available",
                        "Import tooling is disabled in this runtime profile.",
                    )
                    .with_docs_hint(
                        "Set KURA_MCP_ENABLE_IMPORT_PROVIDER_TOOLS=true to re-enable import/provider tools.",
                    ))
                }
            }
            "kura_import_job_get" => {
                if import_device_tools_enabled() {
                    self.tool_import_job_get(args).await
                } else {
                    Err(ToolError::new(
                        "capability_not_available",
                        "Import tooling is disabled in this runtime profile.",
                    )
                    .with_docs_hint(
                        "Set KURA_MCP_ENABLE_IMPORT_PROVIDER_TOOLS=true to re-enable import/provider tools.",
                    ))
                }
            }
            "analyze" => self.tool_analyze(args).await,
            "kura_provider_connections_list" => {
                if import_device_tools_enabled() {
                    self.tool_provider_connections_list(args).await
                } else {
                    Err(ToolError::new(
                        "capability_not_available",
                        "Provider connection tooling is disabled in this runtime profile.",
                    )
                    .with_docs_hint(
                        "Set KURA_MCP_ENABLE_IMPORT_PROVIDER_TOOLS=true to re-enable import/provider tools.",
                    ))
                }
            }
            "kura_provider_connections_upsert" => {
                if import_device_tools_enabled() {
                    self.tool_provider_connections_upsert(args).await
                } else {
                    Err(ToolError::new(
                        "capability_not_available",
                        "Provider connection tooling is disabled in this runtime profile.",
                    )
                    .with_docs_hint(
                        "Set KURA_MCP_ENABLE_IMPORT_PROVIDER_TOOLS=true to re-enable import/provider tools.",
                    ))
                }
            }
            "kura_provider_connection_revoke" => {
                if import_device_tools_enabled() {
                    self.tool_provider_connection_revoke(args).await
                } else {
                    Err(ToolError::new(
                        "capability_not_available",
                        "Provider connection tooling is disabled in this runtime profile.",
                    )
                    .with_docs_hint(
                        "Set KURA_MCP_ENABLE_IMPORT_PROVIDER_TOOLS=true to re-enable import/provider tools.",
                    ))
                }
            }
            "kura_agent_visualization_resolve" => self.tool_agent_visualization_resolve(args).await,
            "kura_observation_draft_dismiss" => self.tool_observation_draft_dismiss(args).await,
            _ => Err(ToolError::new(
                "unknown_tool",
                format!("Unknown tool '{tool_name}'"),
            )),
        }
    }

    async fn tool_api_request(&self, args: &Map<String, Value>) -> Result<Value, ToolError> {
        let method_raw = required_string(args, "method")?;
        let path_raw = required_string(args, "path")?;
        let method = parse_http_method(&method_raw)?;
        let path = normalize_api_path(&path_raw)?;
        if is_mutating_legacy_write_bridge_path(&path, &method) {
            return Err(ToolError::new(
                "legacy_write_bridge_removed",
                "Legacy write bridges are removed in MCP. Do not send routine writes to evidence, maintenance bridges, or resolver detours.",
            )
            .with_field("path")
            .with_docs_hint(
                "This public MCP session is hard-cut to the task-first world. Use record_activity for executed activity logging, plan_activity for future or not-yet-executed workouts, record_recovery/report_health/record_nutrition/record_measurement/record_benchmark for the matching log tasks, and update_profile/update_plan for their matching stateful tasks. For log-style tasks, stay on the same tool with operation=remove|replace when a saved entry must be fixed and omit the target ref only when one saved entry is uniquely determined.",
            ));
        }
        let query = parse_query_pairs_from_args(args)?;
        let body = args.get("body").cloned();
        let include_headers = arg_bool(args, "include_headers", false)?;
        let auth_mode = arg_string(args, "auth_mode", "auto")?;
        let requires_auth = match auth_mode.as_str() {
            "required" => true,
            "none" => false,
            "auto" => path_requires_auth(&path),
            _ => {
                return Err(ToolError::new(
                    "validation_failed",
                    "auth_mode must be one of: auto, required, none",
                )
                .with_field("auth_mode"));
            }
        };

        let response = self
            .send_api_request(
                method.clone(),
                &path,
                &query,
                body,
                requires_auth,
                include_headers,
            )
            .await?;

        Ok(json!({
            "request": {
                "method": method.as_str(),
                "path": path,
                "query": pairs_to_json_object(&query),
                "auth_mode": auth_mode
            },
            "response": response.to_value(),
        }))
    }

    async fn tool_read_query_vnext(&self, args: &Map<String, Value>) -> Result<Value, ToolError> {
        let mut body = Self::build_read_query_vnext_body(args)?;
        let mut payload_budget_retry_applied = false;

        loop {
            let response = self
                .send_api_request(
                    Method::POST,
                    READ_QUERY_V4_PATH,
                    &[],
                    Some(body.clone()),
                    true,
                    false,
                )
                .await?;

            let envelope = build_read_query_tool_envelope(&response, payload_budget_retry_applied);
            let flattened_probe = flatten_public_tool_response(
                "read_query",
                &json!({
                    "status": "complete",
                    "phase": "final",
                    "tool": "read_query",
                    "data": envelope.clone()
                }),
                false,
            );
            if serialized_json_size_bytes(&flattened_probe) <= TOOL_ENVELOPE_MAX_BYTES {
                return Ok(envelope);
            }
            if !Self::shrink_read_query_limit(&mut body) {
                return Ok(envelope);
            }
            payload_budget_retry_applied = true;
        }
    }

    async fn tool_record_recovery_vnext(
        &self,
        args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        let body = Self::build_record_recovery_vnext_body(args)?;
        self.send_task_surface_request("record_recovery", RECORD_RECOVERY_V4_PATH, body)
            .await
    }

    async fn tool_update_profile_vnext(
        &self,
        args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        let body = Self::build_update_profile_vnext_body(args)?;
        self.send_task_surface_request("update_profile", UPDATE_PROFILE_V4_PATH, body)
            .await
    }

    async fn tool_report_health_vnext(
        &self,
        args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        let body = Self::build_report_health_vnext_body(args)?;
        self.send_task_surface_request("report_health", REPORT_HEALTH_V4_PATH, body)
            .await
    }

    async fn tool_record_nutrition_vnext(
        &self,
        args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        let body = Self::build_record_nutrition_vnext_body(args)?;
        self.send_task_surface_request("record_nutrition", RECORD_NUTRITION_V4_PATH, body)
            .await
    }

    async fn tool_record_benchmark_vnext(
        &self,
        args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        let body = Self::build_record_benchmark_vnext_body(args)?;
        self.send_task_surface_request("record_benchmark", RECORD_BENCHMARK_V4_PATH, body)
            .await
    }

    async fn tool_record_measurement_vnext(
        &self,
        args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        let body = Self::build_record_measurement_vnext_body(args)?;
        self.send_task_surface_request("record_measurement", RECORD_MEASUREMENT_V4_PATH, body)
            .await
    }

    async fn tool_update_plan_vnext(&self, args: &Map<String, Value>) -> Result<Value, ToolError> {
        let body = Self::build_update_plan_vnext_body(args)?;
        self.send_task_surface_request("update_plan", UPDATE_PLAN_V4_PATH, body)
            .await
    }

    async fn tool_plan_activity_vnext(
        &self,
        args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        let body = Self::build_plan_activity_vnext_body(args)?;
        self.send_task_surface_request("plan_activity", PLAN_ACTIVITY_V4_PATH, body)
            .await
    }

    async fn send_task_surface_request(
        &self,
        surface: &'static str,
        path: &'static str,
        body: Value,
    ) -> Result<Value, ToolError> {
        let response = self
            .send_api_request(Method::POST, path, &[], Some(body), true, false)
            .await?;

        Ok(json!({
            "request": {
                "path": path,
                "namespace": "task_surface_vnext",
                "surface": surface,
            },
            "response": response.to_value(),
            "contract": {
                "harness_profile": "task_first_vnext"
            }
        }))
    }

    async fn tool_record_activity_vnext(
        &self,
        args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        let body = Self::build_record_activity_vnext_body(args)?;
        let response = self
            .send_api_request(
                Method::POST,
                RECORD_ACTIVITY_V4_PATH,
                &[],
                Some(body),
                true,
                false,
            )
            .await?;

        Ok(json!({
            "request": {
                "path": RECORD_ACTIVITY_V4_PATH,
                "namespace": "task_surface_vnext",
                "surface": "record_activity"
            },
            "response": response.to_value(),
            "contract": {
                "harness_profile": "task_first_vnext"
            }
        }))
    }

    fn clone_public_family_args(
        args: &Map<String, Value>,
        tool_name: &str,
        allowed_fields: &[&str],
    ) -> Result<Map<String, Value>, ToolError> {
        let unknown_fields: Vec<String> = args
            .keys()
            .filter(|key| !allowed_fields.contains(&key.as_str()))
            .cloned()
            .collect();
        if !unknown_fields.is_empty() {
            return Err(ToolError::new(
                "validation_failed",
                format!("Unknown {tool_name} fields: {}", unknown_fields.join(", ")),
            )
            .with_docs_hint(format!(
                "Use only canonical {tool_name} fields documented in the public task contract."
            )));
        }
        Ok(args.clone())
    }

    fn build_record_activity_vnext_body(args: &Map<String, Value>) -> Result<Value, ToolError> {
        Ok(Value::Object(Self::clone_public_family_args(
            args,
            "record_activity",
            &[
                "continuation",
                "date",
                "occurrence_status",
                "blocks",
                "operation",
                "target",
                "replacement_patch",
                "display_label_action",
                "exercise_identity_merge",
                "target_session_id",
                "start_new_session",
                "close_session",
                "session_follow_up",
                "import_assignment",
                "protocol_profile",
                "session_id",
                "timezone",
                "source_kind",
                "idempotency_key",
                "notes",
                "session_context",
                "metadata",
            ],
        )?))
    }

    fn build_record_recovery_vnext_body(args: &Map<String, Value>) -> Result<Value, ToolError> {
        Ok(Value::Object(Self::clone_public_family_args(
            args,
            "record_recovery",
            &[
                "operation",
                "target",
                "continuation",
                "sleep_hours",
                "energy_level",
                "soreness",
                "sleep_quality",
                "sleep_score",
                "quality",
                "readiness_score",
                "bed_time",
                "wake_time",
                "soreness_areas",
                "body_symptoms",
                "notes",
                "bodyweight_kg",
                "motivation",
                "physical_condition",
                "lifestyle_stability",
                "alcohol_last_night",
                "sick_today",
                "training_yesterday",
                "traveling_yesterday",
                "hrv_rmssd",
                "supplements",
                "compact_input",
                "same_day_confirmed_signals",
                "timestamp",
                "idempotency_key",
            ],
        )?))
    }

    fn build_report_health_vnext_body(args: &Map<String, Value>) -> Result<Value, ToolError> {
        Ok(Value::Object(Self::clone_public_family_args(
            args,
            "report_health",
            &[
                "operation",
                "target",
                "kind",
                "affected_area",
                "description",
                "severity",
                "since",
                "label",
                "status",
                "timestamp",
                "idempotency_key",
            ],
        )?))
    }

    fn build_record_nutrition_vnext_body(args: &Map<String, Value>) -> Result<Value, ToolError> {
        Ok(Value::Object(Self::clone_public_family_args(
            args,
            "record_nutrition",
            &[
                "operation",
                "target",
                "kind",
                "description",
                "meal_type",
                "entry_scope",
                "evidence_source",
                "estimate_confidence",
                "portion_confidence",
                "needs_confirmation",
                "assumptions",
                "items",
                "evidence_ref",
                "evidence_notes",
                "calories",
                "protein_g",
                "carbs_g",
                "fat_g",
                "fiber_g",
                "added_sugar_g",
                "saturated_fat_g",
                "sodium_mg",
                "alcohol_units",
                "target_calories",
                "target_protein_g",
                "target_carbs_g",
                "target_fat_g",
                "target_fiber_g",
                "target_added_sugar_g_max",
                "target_saturated_fat_g_max",
                "target_sodium_mg_max",
                "target_alcohol_units_max",
                "timestamp",
                "idempotency_key",
            ],
        )?))
    }

    fn build_record_measurement_vnext_body(args: &Map<String, Value>) -> Result<Value, ToolError> {
        Ok(Value::Object(Self::clone_public_family_args(
            args,
            "record_measurement",
            &[
                "operation",
                "target",
                "kind",
                "weight_kg",
                "body_fat_pct",
                "skeletal_muscle_mass_lbs",
                "time_of_day",
                "conditions",
                "type",
                "value_cm",
                "side",
                "timestamp",
                "idempotency_key",
            ],
        )?))
    }

    fn build_record_benchmark_vnext_body(args: &Map<String, Value>) -> Result<Value, ToolError> {
        Ok(Value::Object(Self::clone_public_family_args(
            args,
            "record_benchmark",
            &[
                "operation",
                "target",
                "test_type",
                "protocol_variant",
                "measured_property",
                "result",
                "attempts",
                "contact_level",
                "auxiliary_metrics",
                "context_text",
                "device_type",
                "surface",
                "session_id",
                "timestamp",
                "idempotency_key",
            ],
        )?))
    }

    fn build_update_profile_vnext_body(args: &Map<String, Value>) -> Result<Value, ToolError> {
        Ok(Value::Object(Self::clone_public_family_args(
            args,
            "update_profile",
            &[
                "preferred_name",
                "experience_level",
                "training_modality",
                "training_frequency_per_week",
                "available_equipment",
                "primary_location",
                "current_program",
                "communication_style",
                "age",
                "date_of_birth",
                "sex",
                "notes",
                "nutrition_approach",
                "nutrition_tracking",
                "primary_adherence_focus",
                "adherence_primary_obstacle",
                "equipment_notes",
                "secondary_location",
                "training_background",
                "goal",
                "goals",
                "preference",
                "preferences",
                "constraints",
                "stature_cm",
                "wingspan_cm",
                "bodyweight_kg",
                "baseline_benchmark",
                "idempotency_key",
            ],
        )?))
    }

    fn build_plan_activity_vnext_body(args: &Map<String, Value>) -> Result<Value, ToolError> {
        Ok(Value::Object(Self::clone_public_family_args(
            args,
            "plan_activity",
            &[
                "date",
                "local_date",
                "exception_id",
                "session_name",
                "name",
                "session_key",
                "blocks",
                "exercises",
                "location",
                "start_time",
                "end_time",
                "notes",
                "note",
                "reason",
                "timezone",
                "session_context",
                "progression_override",
                "timestamp",
                "idempotency_key",
            ],
        )?))
    }

    fn build_update_plan_vnext_body(args: &Map<String, Value>) -> Result<Value, ToolError> {
        Ok(Value::Object(Self::clone_public_family_args(
            args,
            "update_plan",
            &[
                "kind",
                "pending_action_confirmation",
                "read_after_write_targets",
                "name",
                "plan_id",
                "sessions",
                "cycle_weeks",
                "start_date",
                "phase",
                "frequency_per_week",
                "sessions_per_week",
                "progression",
                "notes",
                "change_scope",
                "update_scope",
                "replace_entire_plan",
                "archive_previous_plan",
                "requires_explicit_confirmation",
                "delta",
                "reason",
                "exception_id",
                "operation",
                "selector",
                "target",
                "patch",
                "changes",
                "progression_override",
                "occurrence",
                "scope",
                "expires_at",
                "timestamp",
                "idempotency_key",
            ],
        )?))
    }

    fn build_read_query_vnext_body(args: &Map<String, Value>) -> Result<Value, ToolError> {
        let allowed_fields = [
            "schema_version",
            "source_id",
            "read_kind",
            "subject",
            "filters",
            "query",
            "limit",
            "cursor",
            "output_shape",
            "analysis_mode",
        ];
        let unexpected_fields = args
            .keys()
            .filter(|field| !allowed_fields.contains(&field.as_str()))
            .cloned()
            .collect::<Vec<_>>();
        if !unexpected_fields.is_empty() {
            return Err(
                ToolError::new(
                    "validation_failed",
                    format!("Unknown read_query fields: {}", unexpected_fields.join(", ")),
                )
                .with_field(unexpected_fields[0].clone())
                .with_docs_hint("Use only the bounded read_query fields: source_id, read_kind, and optional subject, filters, query, limit, cursor, output_shape, or analysis_mode."),
            );
        }

        let schema_version = arg_string(args, "schema_version", READ_QUERY_REQUEST_SCHEMA_VERSION)?;
        let source_id = canonical_read_query_source_id(&required_string(args, "source_id")?);
        if matches!(
            source_id.as_str(),
            "authoritative_training_memory" | "training_timeline"
        ) {
            return Err(
                ToolError::new(
                    "validation_failed",
                    "legacy training read sources are removed from the public read_query surface",
                )
                .with_field("source_id")
                .with_docs_hint(
                    "Use source_id=activity_session_history for saved training/activity history, or use analyze for trend and driver questions.",
                ),
            );
        }
        let supported_sources = [
            "activity_session_history",
            "activity_protocol_profiles",
            "training_plan",
            "training_schedule",
            "recovery",
            "performance_tests",
            "measurements",
            "wearable_daily",
            "evidence_fact_health_log",
            "nutrition",
            "supplements",
            "profile",
            "analyze_catalog",
        ];
        if !supported_sources.contains(&source_id.as_str()) {
            return Err(ToolError::new(
                "validation_failed",
                "read_query source_id is not on the current public readable source list",
            )
            .with_field("source_id"));
        }

        let supported_read_kinds: &[&str] = match source_id.as_str() {
            "activity_session_history" => &["lookup", "list", "timeline", "aggregate"],
            "activity_protocol_profiles" => &["lookup", "list", "aggregate"],
            "analyze_catalog" => &["lookup", "list"],
            "training_schedule"
            | "recovery"
            | "performance_tests"
            | "measurements"
            | "wearable_daily"
            | "evidence_fact_health_log"
            | "nutrition"
            | "supplements"
            | "profile" => &["lookup", "list", "timeline", "aggregate"],
            "training_plan" => &["lookup", "list"],
            _ => unreachable!(),
        };

        let read_kind = required_string(args, "read_kind")?;
        if !["lookup", "list", "timeline", "aggregate"].contains(&read_kind.as_str()) {
            return Err(ToolError::new(
                "validation_failed",
                "read_query supports only lookup, list, timeline, or aggregate",
            )
            .with_field("read_kind"));
        }
        if !supported_read_kinds.contains(&read_kind.as_str()) {
            return Err(ToolError::new(
                "validation_failed",
                format!(
                    "this read_query source currently supports {}",
                    supported_read_kinds.join(", ")
                ),
            )
            .with_field("read_kind"));
        }

        let limit = arg_optional_u64(args, "limit")?;
        if limit.is_some_and(|value| !(1..=100).contains(&value)) {
            return Err(ToolError::new(
                "validation_failed",
                "read_query limit must stay between 1 and 100",
            )
            .with_field("limit"));
        }

        let output_shape = arg_optional_string(args, "output_shape")?;
        if let Some(output_shape) = output_shape.as_deref() {
            let allowed = match read_kind.as_str() {
                "lookup" => "item",
                "list" => "collection",
                "timeline" => "series",
                "aggregate" => "aggregate",
                _ => unreachable!(),
            };
            if output_shape != allowed {
                return Err(ToolError::new(
                    "validation_failed",
                    "read_query output_shape does not match read_kind",
                )
                .with_field("output_shape")
                .with_docs_hint(format!(
                    "Use output_shape={allowed} for read_kind={read_kind}."
                )));
            }
        }

        let analysis_mode = arg_optional_string(args, "analysis_mode")?;
        if analysis_mode
            .as_deref()
            .is_some_and(|value| !["none", "allow_hybrid", "require_analysis"].contains(&value))
        {
            return Err(ToolError::new(
                "validation_failed",
                "read_query analysis_mode must be none, allow_hybrid, or require_analysis",
            )
            .with_field("analysis_mode"));
        }

        let mut body = Map::new();
        body.insert("schema_version".to_string(), json!(schema_version));
        body.insert("source_id".to_string(), json!(source_id));
        body.insert("read_kind".to_string(), json!(read_kind));

        if let Some(subject) = arg_optional_string(args, "subject")? {
            body.insert("subject".to_string(), json!(subject));
        }
        if let Some(filters) = args.get("filters") {
            body.insert("filters".to_string(), filters.clone());
        }
        if let Some(query) = args.get("query") {
            body.insert("query".to_string(), query.clone());
        }
        if let Some(limit) = limit {
            body.insert("limit".to_string(), json!(limit));
        }
        if let Some(cursor) = arg_optional_string(args, "cursor")? {
            body.insert("cursor".to_string(), json!(cursor));
        }
        if let Some(output_shape) = output_shape {
            body.insert("output_shape".to_string(), json!(output_shape));
        }
        if let Some(analysis_mode) = analysis_mode {
            body.insert("analysis_mode".to_string(), json!(analysis_mode));
        }

        Ok(Value::Object(body))
    }

    fn shrink_read_query_limit(body: &mut Value) -> bool {
        let Some(object) = body.as_object_mut() else {
            return false;
        };
        let current = object
            .get("limit")
            .and_then(Value::as_u64)
            .unwrap_or(20)
            .clamp(1, 100);
        if current <= 1 {
            return false;
        }
        let next = (current / 2).max(1);
        let next = if next == current { current - 1 } else { next };
        object.insert("limit".to_string(), json!(next));
        true
    }

    async fn tool_events_list(&self, args: &Map<String, Value>) -> Result<Value, ToolError> {
        let event_type = arg_optional_string(args, "event_type")?;
        let since = arg_optional_string(args, "since")?;
        let until = arg_optional_string(args, "until")?;
        let limit = arg_optional_u64(args, "limit")?;
        let cursor = arg_optional_string(args, "cursor")?;
        let auto_paginate = arg_optional_bool(args, "auto_paginate")?.unwrap_or(true);
        let max_pages = arg_optional_u64(args, "max_pages")?
            .unwrap_or(EVENTS_LIST_AUTO_PAGINATE_MAX_PAGES_DEFAULT);
        if max_pages == 0 || max_pages > EVENTS_LIST_AUTO_PAGINATE_MAX_PAGES_MAX {
            return Err(ToolError::new(
                "validation_failed",
                format!(
                    "'max_pages' must be between 1 and {}",
                    EVENTS_LIST_AUTO_PAGINATE_MAX_PAGES_MAX
                ),
            )
            .with_field("max_pages"));
        }
        let max_total = arg_optional_u64(args, "max_total")?
            .unwrap_or(EVENTS_LIST_AUTO_PAGINATE_MAX_TOTAL_DEFAULT);
        if max_total == 0 || max_total > EVENTS_LIST_AUTO_PAGINATE_MAX_TOTAL_MAX {
            return Err(ToolError::new(
                "validation_failed",
                format!(
                    "'max_total' must be between 1 and {}",
                    EVENTS_LIST_AUTO_PAGINATE_MAX_TOTAL_MAX
                ),
            )
            .with_field("max_total"));
        }

        let mut base_query = Vec::new();
        if let Some(value) = event_type.clone() {
            base_query.push(("event_type".to_string(), value));
        }
        if let Some(value) = since.clone() {
            base_query.push(("since".to_string(), value));
        }
        if let Some(value) = until.clone() {
            base_query.push(("until".to_string(), value));
        }
        if let Some(value) = limit {
            base_query.push(("limit".to_string(), value.to_string()));
        }

        let mut query = base_query.clone();
        if let Some(value) = cursor.clone() {
            query.push(("cursor".to_string(), value));
        }

        let response = self
            .send_api_request(Method::GET, "/v1/events", &query, None, true, false)
            .await?;

        let mut response_value = response.to_value();
        let mut event_rows = response
            .body
            .get("data")
            .and_then(Value::as_array)
            .cloned()
            .unwrap_or_default();
        let mut has_more = response
            .body
            .get("has_more")
            .and_then(Value::as_bool)
            .unwrap_or(false);
        let mut next_cursor = response
            .body
            .get("next_cursor")
            .and_then(Value::as_str)
            .map(str::to_string);
        let mut page_count: u64 = 1;
        let mut page_statuses = vec![response.status];
        let auto_paginate_applied = auto_paginate && response.is_success();
        let mut stop_reason = if response.is_success() {
            if auto_paginate_applied {
                "complete".to_string()
            } else {
                "auto_paginate_disabled".to_string()
            }
        } else {
            "initial_http_error".to_string()
        };

        if auto_paginate_applied && has_more {
            let mut seen_cursors = HashSet::<String>::new();
            loop {
                if event_rows.len() >= max_total as usize {
                    event_rows.truncate(max_total as usize);
                    stop_reason = "max_total_reached".to_string();
                    break;
                }
                if page_count >= max_pages {
                    stop_reason = "max_pages_reached".to_string();
                    break;
                }

                let Some(page_cursor) = next_cursor.clone() else {
                    stop_reason = "has_more_missing_next_cursor".to_string();
                    break;
                };

                if !seen_cursors.insert(page_cursor.clone()) {
                    stop_reason = "cursor_cycle_detected".to_string();
                    break;
                }

                let mut page_query = base_query.clone();
                page_query.push(("cursor".to_string(), page_cursor));
                let page_response = self
                    .send_api_request(Method::GET, "/v1/events", &page_query, None, true, false)
                    .await?;
                page_count = page_count.saturating_add(1);
                page_statuses.push(page_response.status);

                if !page_response.is_success() {
                    has_more = true;
                    stop_reason = format!("page_http_error_{}", page_response.status);
                    break;
                }

                let page_rows = page_response
                    .body
                    .get("data")
                    .and_then(Value::as_array)
                    .cloned()
                    .unwrap_or_default();
                event_rows.extend(page_rows);
                has_more = page_response
                    .body
                    .get("has_more")
                    .and_then(Value::as_bool)
                    .unwrap_or(false);
                next_cursor = page_response
                    .body
                    .get("next_cursor")
                    .and_then(Value::as_str)
                    .map(str::to_string);

                if !has_more {
                    stop_reason = "complete".to_string();
                    break;
                }
            }

            if let Some(body) = response_value
                .get_mut("body")
                .and_then(Value::as_object_mut)
            {
                body.insert("data".to_string(), Value::Array(event_rows.clone()));
                body.insert("has_more".to_string(), Value::Bool(has_more));
                body.insert(
                    "next_cursor".to_string(),
                    next_cursor
                        .as_ref()
                        .map(|value| Value::String(value.clone()))
                        .unwrap_or(Value::Null),
                );
            }
        }

        let next_call = build_events_list_pagination_next_call(
            event_type.as_deref(),
            since.as_deref(),
            until.as_deref(),
            limit,
            has_more,
            next_cursor.as_deref(),
        );

        Ok(json!({
            "request": {
                "path": "/v1/events",
                "query": pairs_to_json_object(&query)
            },
            "response": response_value,
            "pagination": {
                "auto_paginate_requested": auto_paginate,
                "auto_paginate_applied": auto_paginate_applied,
                "cursor_supplied": cursor.is_some(),
                "page_count": page_count,
                "total_events": event_rows.len(),
                "has_more": has_more,
                "next_cursor": next_cursor,
                "next_call": next_call,
                "max_pages": max_pages,
                "max_total": max_total,
                "stop_reason": stop_reason,
                "page_statuses": page_statuses
            }
        }))
    }

    async fn tool_projection_get(&self, args: &Map<String, Value>) -> Result<Value, ToolError> {
        let projection_type = required_string(args, "projection_type")?;
        if matches!(
            projection_type.as_str(),
            "authoritative_training_memory" | "training_timeline"
        ) {
            return Err(
                ToolError::new(
                    "validation_failed",
                    "legacy training projections are no longer a public MCP read path",
                )
                .with_field("projection_type")
                .with_docs_hint(
                    "Use read_query with source_id=activity_session_history for saved training/activity history, or use analyze for trend and driver questions.",
                ),
            );
        }
        let key = required_string(args, "key")?;
        let path = format!("/v1/projections/{projection_type}/{key}");

        let response = self
            .send_api_request(Method::GET, &path, &[], None, true, false)
            .await?;

        Ok(json!({
            "request": {
                "path": path,
                "projection_type": projection_type,
                "key": key
            },
            "response": response.to_value()
        }))
    }

    async fn tool_projection_list(&self, args: &Map<String, Value>) -> Result<Value, ToolError> {
        let projection_type = arg_optional_string(args, "projection_type")?;
        let limit = arg_optional_u64(args, "limit")?;
        let cursor = arg_optional_string(args, "cursor")?;
        let mut query = Vec::new();

        let path = if let Some(projection_type) = projection_type {
            if matches!(
                projection_type.as_str(),
                "authoritative_training_memory" | "training_timeline"
            ) {
                return Err(
                    ToolError::new(
                        "validation_failed",
                        "legacy training projections are no longer a public MCP list path",
                    )
                    .with_field("projection_type")
                    .with_docs_hint(
                        "Use read_query with source_id=activity_session_history for saved training/activity history, or use analyze for trend and driver questions.",
                    ),
                );
            }
            if let Some(limit) = limit {
                query.push(("limit".to_string(), limit.to_string()));
            }
            if let Some(cursor) = cursor {
                query.push(("cursor".to_string(), cursor));
            }
            format!("/v1/projections/{projection_type}/paged")
        } else {
            if limit.is_some() || cursor.is_some() {
                return Err(
                    ToolError::new(
                        "validation_failed",
                        "projection_type is required when limit/cursor is provided",
                    )
                    .with_field("projection_type")
                    .with_docs_hint(
                        "Provide projection_type for paged reloads, or omit limit/cursor for full snapshot.",
                    ),
                );
            }
            "/v1/projections".to_string()
        };

        let response = self
            .send_api_request(Method::GET, &path, &query, None, true, false)
            .await?;

        Ok(json!({
            "request": {
                "path": path,
                "query": pairs_to_json_object(&query)
            },
            "response": response.to_value()
        }))
    }

    async fn tool_system_manifest(&self, _args: &Map<String, Value>) -> Result<Value, ToolError> {
        let path = "/v1/system/config/manifest";
        let response = self
            .send_api_request(Method::GET, path, &[], None, true, false)
            .await?;

        Ok(json!({
            "request": { "path": path },
            "response": response.to_value()
        }))
    }

    async fn tool_system_section_get(&self, args: &Map<String, Value>) -> Result<Value, ToolError> {
        let section = required_string(args, "section")?;
        let query = vec![("section".to_string(), section.clone())];
        let path = "/v1/system/config/section";

        let response = self
            .send_api_request(Method::GET, path, &query, None, true, false)
            .await?;

        Ok(json!({
            "request": {
                "path": path,
                "query": pairs_to_json_object(&query)
            },
            "response": response.to_value()
        }))
    }

    async fn tool_agent_answer_admissibility(
        &self,
        args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        let task_intent = required_string(args, "task_intent")?;
        let draft_answer = required_string(args, "draft_answer")?;
        let used_sources = arg_optional_string_array(args, "used_sources")?;
        let path = "/v1/agent/answer-admissibility";
        let response = self
            .send_api_request(
                Method::POST,
                path,
                &[],
                Some(json!({
                    "task_intent": task_intent,
                    "draft_answer": draft_answer,
                    "used_sources": used_sources,
                })),
                true,
                false,
            )
            .await?;
        Ok(json!({
            "request": { "path": path },
            "response": response.to_value()
        }))
    }

    async fn tool_semantic_resolve(&self, args: &Map<String, Value>) -> Result<Value, ToolError> {
        let queries = args
            .get("queries")
            .and_then(Value::as_array)
            .ok_or_else(|| {
                ToolError::new("validation_failed", "Missing required field 'queries'")
                    .with_field("queries")
            })?;
        if queries.is_empty() {
            return Err(
                ToolError::new("validation_failed", "'queries' must not be empty")
                    .with_field("queries"),
            );
        }

        let mut body = json!({
            "queries": queries
        });
        if let Some(top_k) = arg_optional_u64(args, "top_k")? {
            body["top_k"] = json!(top_k);
        }

        let response = self
            .send_api_request(
                Method::POST,
                "/v1/semantic/resolve",
                &[],
                Some(body),
                true,
                false,
            )
            .await?;

        Ok(json!({
            "request": {
                "path": "/v1/semantic/resolve",
                "query_count": queries.len()
            },
            "response": response.to_value()
        }))
    }

    async fn tool_exercise_vocabulary_discover(
        &self,
        args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        let mut query = vec![("domain".to_string(), "exercise".to_string())];
        let mut request_query = json!({ "domain": "exercise" });

        if let Some(raw_query) = arg_optional_string(args, "query")? {
            let normalized = raw_query.trim();
            if !normalized.is_empty() {
                query.push(("query".to_string(), normalized.to_string()));
                request_query["query"] = json!(normalized);
            }
        }

        if let Some(limit) = arg_optional_u64(args, "limit")? {
            if !(1..=200).contains(&limit) {
                return Err(ToolError::new(
                    "validation_failed",
                    "'limit' must be between 1 and 200",
                )
                .with_field("limit"));
            }
            query.push(("limit".to_string(), limit.to_string()));
            request_query["limit"] = json!(limit);
        }

        let response = self
            .send_api_request(
                Method::GET,
                "/v1/semantic/catalog",
                &query,
                None,
                true,
                false,
            )
            .await?;

        Ok(json!({
            "request": {
                "path": "/v1/semantic/catalog",
                "query": request_query
            },
            "response": response.to_value()
        }))
    }

    async fn tool_access_request(&self, args: &Map<String, Value>) -> Result<Value, ToolError> {
        let email = required_string(args, "email")?;
        let mut body = json!({ "email": email });
        if let Some(name) = arg_optional_string(args, "name")? {
            body["name"] = json!(name);
        }
        if let Some(context) = arg_optional_string(args, "context")? {
            body["context"] = json!(context);
        }
        if let Some(locale) = arg_optional_string(args, "locale")? {
            body["locale"] = json!(locale);
        }
        if let Some(turnstile_token) = arg_optional_string(args, "turnstile_token")? {
            body["turnstile_token"] = json!(turnstile_token);
        }

        let response = self
            .send_api_request(
                Method::POST,
                "/v1/access/request",
                &[],
                Some(body),
                false,
                false,
            )
            .await?;

        Ok(json!({
            "request": { "path": "/v1/access/request" },
            "response": response.to_value()
        }))
    }

    async fn tool_account_api_keys_list(
        &self,
        _args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        let response = self
            .send_api_request(Method::GET, "/v1/account/api-keys", &[], None, true, false)
            .await?;

        Ok(json!({
            "request": { "path": "/v1/account/api-keys" },
            "response": response.to_value()
        }))
    }

    async fn tool_account_api_keys_create(
        &self,
        args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        let label = required_string(args, "label")?;
        let mut body = json!({ "label": label });
        if let Some(scopes) = arg_optional_string_array(args, "scopes")? {
            body["scopes"] = json!(scopes);
        }

        let response = self
            .send_api_request(
                Method::POST,
                "/v1/account/api-keys",
                &[],
                Some(body),
                true,
                false,
            )
            .await?;

        Ok(json!({
            "request": { "path": "/v1/account/api-keys" },
            "response": response.to_value()
        }))
    }

    async fn tool_account_api_keys_revoke(
        &self,
        args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        let key_id = required_string(args, "key_id")?;
        let key_id = parse_uuid_string(&key_id, "key_id")?;
        let path = format!("/v1/account/api-keys/{key_id}");

        let response = self
            .send_api_request(Method::DELETE, &path, &[], None, true, false)
            .await?;

        Ok(json!({
            "request": { "path": path, "key_id": key_id },
            "response": response.to_value()
        }))
    }

    async fn tool_import_job_create(&self, args: &Map<String, Value>) -> Result<Value, ToolError> {
        let provider = required_string(args, "provider")?;
        let provider_user_id = required_string(args, "provider_user_id")?;
        let file_format = required_string(args, "file_format")?;
        let payload_text = required_string(args, "payload_text")?;
        let external_activity_id = required_string(args, "external_activity_id")?;

        let mut body = json!({
            "provider": provider,
            "provider_user_id": provider_user_id,
            "file_format": file_format,
            "payload_text": payload_text,
            "external_activity_id": external_activity_id
        });
        if let Some(external_event_version) = arg_optional_string(args, "external_event_version")? {
            body["external_event_version"] = json!(external_event_version);
        }
        if let Some(raw_payload_ref) = arg_optional_string(args, "raw_payload_ref")? {
            body["raw_payload_ref"] = json!(raw_payload_ref);
        }
        if let Some(ingestion_method) = arg_optional_string(args, "ingestion_method")? {
            body["ingestion_method"] = json!(ingestion_method);
        }

        let response = self
            .send_api_request(
                Method::POST,
                "/v1/imports/jobs",
                &[],
                Some(body),
                true,
                false,
            )
            .await?;

        Ok(json!({
            "request": { "path": "/v1/imports/jobs" },
            "response": response.to_value()
        }))
    }

    async fn tool_import_job_get(&self, args: &Map<String, Value>) -> Result<Value, ToolError> {
        let job_id = required_string(args, "job_id")?;
        let job_id = parse_uuid_string(&job_id, "job_id")?;
        let path = format!("/v1/imports/jobs/{job_id}");

        let response = self
            .send_api_request(Method::GET, &path, &[], None, true, false)
            .await?;

        Ok(json!({
            "request": { "path": path, "job_id": job_id },
            "response": response.to_value()
        }))
    }

    async fn tool_analyze(&self, args: &Map<String, Value>) -> Result<Value, ToolError> {
        let mut body = build_analyze_body(args)?;
        prepare_agent_analyze_body(&mut body);
        let response = self
            .send_api_request(Method::POST, ANALYZE_ENDPOINT, &[], Some(body), true, false)
            .await?;

        Ok(json!({
            "request": { "path": ANALYZE_ENDPOINT },
            "response": response.to_value()
        }))
    }

    async fn tool_provider_connections_list(
        &self,
        _args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        let response = self
            .send_api_request(
                Method::GET,
                "/v1/providers/connections",
                &[],
                None,
                true,
                false,
            )
            .await?;

        Ok(json!({
            "request": { "path": "/v1/providers/connections" },
            "response": response.to_value()
        }))
    }

    async fn tool_provider_connections_upsert(
        &self,
        args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        let provider = required_string(args, "provider")?;
        let provider_account_id = required_string(args, "provider_account_id")?;
        let auth_state = required_string(args, "auth_state")?;

        let mut body = json!({
            "provider": provider,
            "provider_account_id": provider_account_id,
            "auth_state": auth_state,
        });
        if let Some(scopes) = arg_optional_string_array(args, "scopes")? {
            body["scopes"] = json!(scopes);
        }
        if let Some(consented_at) = arg_optional_string(args, "consented_at")? {
            body["consented_at"] = json!(consented_at);
        }
        if let Some(token_expires_at) = arg_optional_string(args, "token_expires_at")? {
            body["token_expires_at"] = json!(token_expires_at);
        }
        if let Some(sync_cursor) = arg_optional_string(args, "sync_cursor")? {
            body["sync_cursor"] = json!(sync_cursor);
        }
        if let Some(access_token_ref) = arg_optional_string(args, "access_token_ref")? {
            body["access_token_ref"] = json!(access_token_ref);
        }
        if let Some(refresh_token_ref) = arg_optional_string(args, "refresh_token_ref")? {
            body["refresh_token_ref"] = json!(refresh_token_ref);
        }
        if let Some(token_fingerprint) = arg_optional_string(args, "token_fingerprint")? {
            body["token_fingerprint"] = json!(token_fingerprint);
        }
        if let Some(last_oauth_state_nonce) = arg_optional_string(args, "last_oauth_state_nonce")? {
            body["last_oauth_state_nonce"] = json!(last_oauth_state_nonce);
        }
        if let Some(last_error_code) = arg_optional_string(args, "last_error_code")? {
            body["last_error_code"] = json!(last_error_code);
        }
        if let Some(last_error_at) = arg_optional_string(args, "last_error_at")? {
            body["last_error_at"] = json!(last_error_at);
        }

        let response = self
            .send_api_request(
                Method::POST,
                "/v1/providers/connections",
                &[],
                Some(body),
                true,
                false,
            )
            .await?;

        Ok(json!({
            "request": { "path": "/v1/providers/connections" },
            "response": response.to_value()
        }))
    }

    async fn tool_provider_connection_revoke(
        &self,
        args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        let connection_id = required_string(args, "connection_id")?;
        let connection_id = parse_uuid_string(&connection_id, "connection_id")?;
        let reason = required_string(args, "reason")?;
        let path = format!("/v1/providers/connections/{connection_id}/revoke");
        let body = json!({ "reason": reason });

        let response = self
            .send_api_request(Method::POST, &path, &[], Some(body), true, false)
            .await?;

        Ok(json!({
            "request": { "path": path, "connection_id": connection_id },
            "response": response.to_value()
        }))
    }

    async fn tool_agent_visualization_resolve(
        &self,
        args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        let task_intent = required_string(args, "task_intent")?;
        let mut body = json!({
            "task_intent": task_intent,
            "allow_rich_rendering": arg_bool(args, "allow_rich_rendering", true)?
        });
        if let Some(value) = arg_optional_string(args, "user_preference_override")? {
            body["user_preference_override"] = json!(value);
        }
        if let Some(value) = arg_optional_string(args, "complexity_hint")? {
            body["complexity_hint"] = json!(value);
        }
        if let Some(value) = arg_optional_string(args, "telemetry_session_id")? {
            body["telemetry_session_id"] = json!(value);
        }
        if let Some(spec) = args.get("visualization_spec") {
            if !spec.is_object() {
                return Err(ToolError::new(
                    "validation_failed",
                    "visualization_spec must be an object when provided",
                )
                .with_field("visualization_spec"));
            }
            body["visualization_spec"] = spec.clone();
        }

        let response = self
            .send_api_request(
                Method::POST,
                "/v1/agent/visualization/resolve",
                &[],
                Some(body),
                true,
                false,
            )
            .await?;

        Ok(json!({
            "request": { "path": "/v1/agent/visualization/resolve" },
            "response": response.to_value()
        }))
    }

    async fn tool_observation_draft_dismiss(
        &self,
        args: &Map<String, Value>,
    ) -> Result<Value, ToolError> {
        let observation_id = required_string(args, "observation_id")?;
        let observation_id = parse_uuid_string(&observation_id, "observation_id")?;
        let path = format!("/v1/agent/observation-drafts/{observation_id}/dismiss");

        let mut body = json!({});
        if let Some(reason) = arg_optional_string(args, "reason")? {
            body["reason"] = json!(reason);
        }
        if let Some(source) = arg_optional_string(args, "source")? {
            body["source"] = json!(source);
        }
        if let Some(agent) = arg_optional_string(args, "agent")? {
            body["agent"] = json!(agent);
        }
        if let Some(device) = arg_optional_string(args, "device")? {
            body["device"] = json!(device);
        }
        if let Some(session_id) = arg_optional_string(args, "session_id")? {
            body["session_id"] = json!(session_id);
        }
        if let Some(idempotency_key) = arg_optional_string(args, "idempotency_key")? {
            body["idempotency_key"] = json!(idempotency_key);
        }

        let has_body = body
            .as_object()
            .map(|payload| !payload.is_empty())
            .unwrap_or(false);
        let body = if has_body { Some(body) } else { None };

        let response = self
            .send_api_request(Method::POST, &path, &[], body, true, false)
            .await?;

        Ok(json!({
            "request": { "path": path, "observation_id": observation_id },
            "response": response.to_value()
        }))
    }

    fn resources_list_payload(&self) -> Value {
        let resources: Vec<Value> = resource_definitions()
            .into_iter()
            .map(|res| {
                json!({
                    "uri": res.uri,
                    "name": res.name,
                    "description": res.description,
                    "mimeType": "application/json"
                })
            })
            .collect();
        json!({ "resources": resources })
    }

    async fn handle_resources_read(&self, params: Value) -> Result<Value, RpcError> {
        let params = params
            .as_object()
            .ok_or_else(|| RpcError::invalid_params("resources/read params must be an object"))?;
        let uri = params.get("uri").and_then(Value::as_str).ok_or_else(|| {
            RpcError::invalid_params("resources/read requires string field 'uri'")
        })?;

        let content_payload = match uri {
            "kura://openapi" => self
                .send_api_request(
                    Method::GET,
                    "/api-doc/openapi.json",
                    &[],
                    None,
                    false,
                    false,
                )
                .await
                .map(|r| r.to_value())
                .map_err(|e| RpcError::internal(e.message))?,
            "kura://agent/capabilities" => self
                .send_api_request(
                    Method::GET,
                    "/v1/agent/capabilities",
                    &[],
                    None,
                    true,
                    false,
                )
                .await
                .map(|r| r.to_value())
                .map_err(|e| RpcError::internal(e.message))?,
            "kura://agent/skillpacks" => self
                .send_api_request(
                    Method::GET,
                    AGENT_SKILLPACK_BRIEF_PATH,
                    &[("surface".to_string(), "mcp".to_string())],
                    None,
                    true,
                    false,
                )
                .await
                .map(|r| r.to_value())
                .map_err(|e| RpcError::internal(e.message))?,
            "kura://system/config" => self
                .send_api_request(Method::GET, "/v1/system/config", &[], None, true, false)
                .await
                .map(|r| r.to_value())
                .map_err(|e| RpcError::internal(e.message))?,
            "kura://system/config/manifest" => self
                .send_api_request(
                    Method::GET,
                    "/v1/system/config/manifest",
                    &[],
                    None,
                    true,
                    false,
                )
                .await
                .map(|r| r.to_value())
                .map_err(|e| RpcError::internal(e.message))?,
            _ => {
                return Err(RpcError::invalid_params(format!(
                    "Unknown resource uri '{uri}'"
                )));
            }
        };

        Ok(json!({
            "contents": [{
                "uri": uri,
                "mimeType": "application/json",
                "text": to_pretty_json(&content_payload)
            }]
        }))
    }

    async fn resolve_session_identity_for_rollout(&self) {
        if let Some((user_id, email)) = get_cached_session_identity(&self.session_id) {
            record_session_identity_for_rollout(&self.session_id, user_id, email);
            return;
        }

        let (user_id, email) = match self
            .send_api_request(Method::GET, "/v1/auth/me", &[], None, true, false)
            .await
        {
            Ok(result) if result.status < 400 => extract_auth_me_identity(&result.body),
            _ => (None, None),
        };

        cache_session_identity(&self.session_id, user_id.clone(), email.clone());
        record_session_identity_for_rollout(&self.session_id, user_id, email);
    }

    async fn send_api_request(
        &self,
        method: Method,
        path: &str,
        query: &[(String, String)],
        body: Option<Value>,
        requires_auth: bool,
        include_headers: bool,
    ) -> Result<ApiCallResult, ToolError> {
        let path = normalize_api_path(path)?;
        if is_admin_api_path(&path) && !self.config.allow_admin {
            return Err(
                ToolError::new(
                    "admin_path_blocked",
                    "Admin API paths are disabled in MCP by default",
                )
                .with_field("path")
                .with_docs_hint(
                    "Start MCP with --allow-admin (or set KURA_MCP_ALLOW_ADMIN=1) only in trusted developer/admin sessions.",
                ),
            );
        }
        let mut url = reqwest::Url::parse(&format!(
            "{}{}",
            self.config.api_url.trim_end_matches('/'),
            path
        ))
        .map_err(|e| ToolError::new("invalid_url", format!("Invalid API URL/path: {e}")))?;
        if !query.is_empty() {
            let mut qp = url.query_pairs_mut();
            for (k, v) in query {
                qp.append_pair(k, v);
            }
        }

        let mut request = self.http.request(method, url);
        request = request.header(KURA_CLIENT_NAME_HEADER, MCP_SERVER_NAME);
        request = request.header(KURA_CLIENT_VERSION_HEADER, env!("CARGO_PKG_VERSION"));
        request = request.header(
            KURA_CLIENT_INSTALL_CHANNEL_HEADER,
            mcp_install_channel_header_value(),
        );
        if let Some(ack_header_value) = pending_notice_ack_header_value(&self.session_id) {
            request = request.header(KURA_CLIENT_NOTICE_ACK_HEADER, ack_header_value);
        }
        if requires_auth && !self.config.no_auth {
            let token = self.resolve_bearer_token().await?;
            request = request.header("Authorization", format!("Bearer {token}"));
        }
        if let Some(body) = body {
            request = request.json(&body);
        }

        let response = request.send().await.map_err(|e| {
            ToolError::new(
                "connection_error",
                format!("Failed to reach Kura API at {}: {e}", self.config.api_url),
            )
            .with_docs_hint("Ensure the API is running and KURA_API_URL points to it.")
        })?;

        let status = response.status().as_u16();
        let headers = if include_headers {
            Some(
                response
                    .headers()
                    .iter()
                    .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("<binary>").to_string()))
                    .collect::<BTreeMap<_, _>>(),
            )
        } else {
            None
        };
        let bytes = response.bytes().await.map_err(|e| {
            ToolError::new(
                "response_error",
                format!("Failed to read API response body: {e}"),
            )
        })?;
        let body = parse_response_body(&bytes);
        if (200..=299).contains(&status) {
            queue_pending_notice_ack_ids(
                &self.session_id,
                extract_agent_user_notice_ack_ids(Some(&body)),
            );
        }

        Ok(ApiCallResult {
            status,
            body,
            headers,
        })
    }

    async fn resolve_bearer_token(&self) -> Result<String, ToolError> {
        if let Some(token) = &self.config.explicit_token {
            return Ok(token.clone());
        }
        resolve_token(&self.config.api_url).await.map_err(|e| {
            ToolError::new("auth_missing", e.to_string())
                .with_docs_hint("Run `kura login`, set KURA_API_KEY, or pass --token.")
        })
    }
}

#[derive(Debug)]
struct RpcError {
    code: i64,
    message: String,
    data: Option<Value>,
}

impl RpcError {
    fn invalid_request(message: impl Into<String>) -> Self {
        Self {
            code: -32600,
            message: message.into(),
            data: None,
        }
    }

    fn method_not_found(method: &str) -> Self {
        Self {
            code: -32601,
            message: format!("Method not found: {method}"),
            data: None,
        }
    }

    fn invalid_params(message: impl Into<String>) -> Self {
        Self {
            code: -32602,
            message: message.into(),
            data: None,
        }
    }

    fn internal(message: impl Into<String>) -> Self {
        Self {
            code: -32603,
            message: message.into(),
            data: None,
        }
    }
}

#[derive(Debug, Clone)]
struct ToolError {
    code: String,
    message: String,
    field: Option<String>,
    docs_hint: Option<String>,
    next_action: Option<String>,
    next_action_url: Option<String>,
    details: Option<Value>,
}

impl ToolError {
    fn new(code: impl Into<String>, message: impl Into<String>) -> Self {
        Self {
            code: code.into(),
            message: message.into(),
            field: None,
            docs_hint: None,
            next_action: None,
            next_action_url: None,
            details: None,
        }
    }

    fn with_field(mut self, field: impl Into<String>) -> Self {
        self.field = Some(field.into());
        self
    }

    fn with_docs_hint(mut self, docs_hint: impl Into<String>) -> Self {
        self.docs_hint = Some(docs_hint.into());
        self
    }

    #[cfg(test)]
    fn with_next_action(mut self, next_action: impl Into<String>) -> Self {
        self.next_action = Some(next_action.into());
        self
    }

    #[cfg(test)]
    fn with_next_action_url(mut self, next_action_url: impl Into<String>) -> Self {
        self.next_action_url = Some(next_action_url.into());
        self
    }

    #[cfg(test)]
    fn with_details(mut self, details: Value) -> Self {
        self.details = Some(details);
        self
    }

    fn to_value(&self) -> Value {
        let mut payload = json!({
            "error": self.code,
            "message": self.message
        });
        if let Some(field) = &self.field {
            payload["field"] = Value::String(field.clone());
        }
        if let Some(docs_hint) = &self.docs_hint {
            payload["docs_hint"] = Value::String(docs_hint.clone());
        }
        if let Some(next_action) = &self.next_action {
            payload["next_action"] = Value::String(next_action.clone());
        }
        if let Some(next_action_url) = &self.next_action_url {
            payload["next_action_url"] = Value::String(next_action_url.clone());
        }
        if let Some(details) = &self.details {
            payload["details"] = details.clone();
        }
        payload
    }
}

#[derive(Debug)]
struct ApiCallResult {
    status: u16,
    body: Value,
    headers: Option<BTreeMap<String, String>>,
}

impl ApiCallResult {
    fn is_success(&self) -> bool {
        (200..=299).contains(&self.status)
    }

    fn to_value(&self) -> Value {
        let mut value = json!({
            "ok": self.is_success(),
            "status": self.status,
            "body": self.body
        });
        if let Some(headers) = &self.headers {
            value["headers"] = json!(headers);
        }
        value
    }
}

#[cfg(test)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum WriteMode {
    Commit,
    FamilyWrite,
}

#[cfg(test)]
impl WriteMode {
    fn as_str(self) -> &'static str {
        match self {
            WriteMode::Commit => "commit",
            WriteMode::FamilyWrite => "family_write",
        }
    }
}

#[derive(Debug, Clone)]
struct ToolDefinition {
    name: &'static str,
    description: &'static str,
    input_schema: Value,
}

#[derive(Debug)]
struct ResourceDefinition {
    uri: &'static str,
    name: &'static str,
    description: &'static str,
}

fn vnext_public_surface_tool_names() -> &'static [&'static str] {
    &[
        "read_query",
        "analyze",
        "record_activity",
        "record_recovery",
        "report_health",
        "record_nutrition",
        "record_measurement",
        "record_benchmark",
        "update_profile",
        "plan_activity",
        "update_plan",
    ]
}

fn is_tool_exposed_on_public_surface(name: &str) -> bool {
    vnext_public_surface_tool_names().contains(&name)
}

fn published_tool_definitions() -> Vec<ToolDefinition> {
    vnext_tool_definitions()
        .into_iter()
        .filter(|tool| is_tool_exposed_on_public_surface(tool.name))
        .map(|tool| ToolDefinition {
            input_schema: sanitize_tool_input_schema(&tool.input_schema),
            ..tool
        })
        .collect()
}

#[cfg(test)]
fn tool_definitions() -> Vec<ToolDefinition> {
    let mut tools = vec![
        ToolDefinition {
            name: "kura_api_request",
            description: "Generic API request fallback for non-hardcoded workflows.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "method": { "type": "string", "enum": ["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"] },
                    "path": { "type": "string", "description": "Absolute API path like /v1/events" },
                    "query": {
                        "type": "object",
                        "description": "Query string as an object map, for example {\"limit\":10}. Use query_entries when your client prefers explicit key/value rows.",
                        "additionalProperties": {
                            "description": "Scalar query value (string, number, or boolean)."
                        }
                    },
                    "query_entries": {
                        "type": "array",
                        "description": "Compatibility alias for query as [{\"key\":\"limit\",\"value\":10}]. Use either query or query_entries, not both.",
                        "items": {
                            "type": "object",
                            "properties": {
                                "key": { "type": "string" },
                                "value": {
                                    "description": "Scalar query value (string, number, or boolean)."
                                }
                            },
                            "required": ["key", "value"],
                            "additionalProperties": false
                        }
                    },
                    "body": {},
                    "auth_mode": { "type": "string", "enum": ["auto", "required", "none"], "default": "auto" },
                    "include_headers": { "type": "boolean", "default": false }
                },
                "required": ["method", "path"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "kura_events_list",
            description: "List events with cursor/time/type filters. Auto-pagination is enabled by default with bounded safety caps.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "event_type": { "type": "string" },
                    "since": { "type": "string", "description": "RFC3339 timestamp inclusive" },
                    "until": { "type": "string", "description": "RFC3339 timestamp exclusive" },
                    "limit": { "type": "integer", "minimum": 1, "maximum": 200 },
                    "cursor": { "type": "string" },
                    "auto_paginate": {
                        "type": "boolean",
                        "default": true,
                        "description": "When true, follows next_cursor until completion or cap."
                    },
                    "max_pages": {
                        "type": "integer",
                        "minimum": 1,
                        "maximum": 50,
                        "default": 8,
                        "description": "Safety cap for auto-pagination page count."
                    },
                    "max_total": {
                        "type": "integer",
                        "minimum": 1,
                        "maximum": 20000,
                        "default": 1000,
                        "description": "Safety cap for total returned events after auto-pagination."
                    }
                },
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "kura_projection_get",
            description: "Read one projection by type/key.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "projection_type": { "type": "string" },
                    "key": { "type": "string" }
                },
                "required": ["projection_type", "key"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "kura_projection_list",
            description: "List projections by type or full snapshot.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "projection_type": { "type": "string" },
                    "limit": {
                        "type": "integer",
                        "minimum": 1,
                        "maximum": 200,
                        "description": "Only with projection_type. Uses paged projection contract."
                    },
                    "cursor": {
                        "type": "string",
                        "description": "Only with projection_type. Opaque cursor from previous paged response."
                    }
                },
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "kura_system_manifest",
            description: "Read machine-readable system section manifest. Use kura_system_manifest to list sections before targeted reloads.",
            input_schema: json!({
                "type": "object",
                "properties": {},
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "kura_system_section_get",
            description: "Read one system config section by manifest section id.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "section": { "type": "string" }
                },
                "required": ["section"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "kura_agent_answer_admissibility",
            description: "Validate a draft coaching or analytical answer against Kura's authoritative serving view and provenance contract before replying. Validate any named session with kura_agent_answer_admissibility before replying.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "task_intent": { "type": "string", "description": "Pass the current user request verbatim when possible." },
                    "draft_answer": { "type": "string", "description": "Draft user-facing answer to validate." },
                    "used_sources": {
                        "type": "array",
                        "description": "Canonical Kura surfaces actually used while drafting the answer, e.g. performance_tests.aggregate or strength_inference.front_squat.",
                        "items": { "type": "string" }
                    }
                },
                "required": ["task_intent", "draft_answer"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "kura_semantic_resolve",
            description: "Resolve free-text exercise/food terms to canonical keys.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "queries": {
                        "type": "array",
                        "minItems": 1,
                        "items": {
                            "type": "object",
                            "properties": {
                                "term": { "type": "string" },
                                "domain": { "type": "string", "enum": ["exercise", "food"] }
                            },
                            "required": ["term", "domain"],
                            "additionalProperties": false
                        }
                    },
                    "top_k": { "type": "integer", "minimum": 1, "maximum": 10 }
                },
                "required": ["queries"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "kura_exercise_vocabulary_discover",
            description: "Discover exercise vocabulary entries from semantic catalog (canonical ids + variants).",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "query": { "type": "string", "description": "Optional case-insensitive filter over key/label/variants" },
                    "limit": { "type": "integer", "minimum": 1, "maximum": 200 }
                },
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "kura_access_request",
            description: "Submit a public access request (no auth required).",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "email": { "type": "string" },
                    "name": { "type": "string" },
                    "context": { "type": "string" },
                    "locale": { "type": "string", "enum": ["de", "en", "ja"] },
                    "turnstile_token": { "type": "string" }
                },
                "required": ["email"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "kura_account_api_keys_list",
            description: "List API keys for the authenticated account.",
            input_schema: json!({
                "type": "object",
                "properties": {},
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "kura_account_api_keys_create",
            description: "Create an API key for the authenticated account.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "label": { "type": "string" },
                    "scopes": {
                        "type": "array",
                        "items": { "type": "string" }
                    }
                },
                "required": ["label"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "kura_account_api_keys_revoke",
            description: "Revoke an API key by key_id.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "key_id": { "type": "string", "description": "UUID" }
                },
                "required": ["key_id"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "kura_import_job_create",
            description: "Queue a new external import job.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "provider": { "type": "string" },
                    "provider_user_id": { "type": "string" },
                    "file_format": { "type": "string" },
                    "payload_text": { "type": "string" },
                    "external_activity_id": { "type": "string" },
                    "external_event_version": { "type": "string" },
                    "raw_payload_ref": { "type": "string" },
                    "ingestion_method": { "type": "string" }
                },
                "required": [
                    "provider",
                    "provider_user_id",
                    "file_format",
                    "payload_text",
                    "external_activity_id"
                ],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "kura_import_job_get",
            description: "Get status of an import job by job_id.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "job_id": { "type": "string", "description": "UUID" }
                },
                "required": ["job_id"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "analyze",
            description: analyze_tool_description(),
            input_schema: analyze_input_schema(),
        },
        ToolDefinition {
            name: "kura_provider_connections_list",
            description: "List provider connections for the authenticated account.",
            input_schema: json!({
                "type": "object",
                "properties": {},
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "kura_provider_connections_upsert",
            description: "Upsert provider connection metadata.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "provider": { "type": "string" },
                    "provider_account_id": { "type": "string" },
                    "auth_state": { "type": "string" },
                    "scopes": {
                        "type": "array",
                        "items": { "type": "string" }
                    },
                    "consented_at": { "type": "string", "description": "RFC3339 timestamp" },
                    "token_expires_at": { "type": "string", "description": "RFC3339 timestamp" },
                    "sync_cursor": { "type": "string" },
                    "access_token_ref": { "type": "string" },
                    "refresh_token_ref": { "type": "string" },
                    "token_fingerprint": { "type": "string" },
                    "last_oauth_state_nonce": { "type": "string" },
                    "last_error_code": { "type": "string" },
                    "last_error_at": { "type": "string", "description": "RFC3339 timestamp" }
                },
                "required": ["provider", "provider_account_id", "auth_state"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "kura_provider_connection_revoke",
            description: "Revoke one provider connection by connection_id.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "connection_id": { "type": "string", "description": "UUID" },
                    "reason": { "type": "string" }
                },
                "required": ["connection_id", "reason"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "kura_agent_visualization_resolve",
            description: "Resolve visualization policy and output for a task intent.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "task_intent": { "type": "string" },
                    "user_preference_override": { "type": "string", "enum": ["auto", "always", "never"] },
                    "complexity_hint": { "type": "string", "enum": ["low", "medium", "high"] },
                    "allow_rich_rendering": { "type": "boolean", "default": true },
                    "visualization_spec": {
                        "type": "object",
                        "description": "AgentVisualizationSpec payload"
                    },
                    "telemetry_session_id": { "type": "string" }
                },
                "required": ["task_intent"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "kura_observation_draft_dismiss",
            description: "Dismiss one observation draft (duplicate/test/noise) and retract it.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "observation_id": { "type": "string", "description": "UUID" },
                    "reason": { "type": "string" },
                    "source": { "type": "string" },
                    "agent": { "type": "string" },
                    "device": { "type": "string" },
                    "session_id": { "type": "string" },
                    "idempotency_key": { "type": "string" }
                },
                "required": ["observation_id"],
                "additionalProperties": false
            }),
        },
    ];
    if !import_device_tools_enabled() {
        tools.retain(|tool| {
            !matches!(
                tool.name,
                "kura_import_job_create"
                    | "kura_import_job_get"
                    | "kura_provider_connections_list"
                    | "kura_provider_connections_upsert"
                    | "kura_provider_connection_revoke"
            )
        });
    }
    tools
}

fn canonical_read_query_source_id(source_id: &str) -> String {
    match source_id
        .trim()
        .to_ascii_lowercase()
        .replace('-', "_")
        .replace(' ', "_")
        .as_str()
    {
        "benchmark"
        | "benchmarks"
        | "benchmark_results"
        | "performance_test"
        | "performance_test_truth"
        | "jump_tests" => "performance_tests".to_string(),
        "wearable_daily_truth" => "wearable_daily".to_string(),
        _ => source_id.trim().to_string(),
    }
}

fn read_query_vnext_schema() -> Value {
    json!({
        "type": "object",
        "description": "Universal canonical read surface on /v4/agent/read-query. Use this for facts, bounded summaries, exact records, and raw timelines across the live canonical sources on this branch. Use analyze first for trend, stagnation, driver, influence, relationship, cross-domain explanation, comparison, or training-decision questions. Do not use read_query or analyze_catalog as routine preflight for analysis questions; analyze resolves semantic requests directly. Activity reads expose comparison_stream as the exact-progress grouping and exercise_identity as a family/mixed-variant view. Benchmark, jump-test, or performance-test history questions should use canonical source_id=performance_tests; common aliases such as benchmarks are normalized to performance_tests. Imported daily wearable metrics such as steps, resting heart rate, stress, Body Battery, respiration, and Pulse Ox use source_id=wearable_daily. Use source_id=analyze_catalog only after analyze asks for catalog lookup, or when the user explicitly asks what observed analysis vocabulary exists. subject=candidates returns the paged observed candidate catalog with opaque handles and analysis_variable_template fills; subject=vocabulary_search returns compact ranked candidates with cursor continuation.",
        "examples": [
            {
                "label": "compact activity history for recent days",
                "payload": {
                    "source_id": "activity_session_history",
                    "read_kind": "list",
                    "subject": "day_history",
                    "filters": { "date_from": "2026-04-02", "date_to": "2026-04-08" },
                    "output_shape": "collection"
                }
            },
            {
                "label": "exact session detail with cursor-paged blocks",
                "payload": {
                    "source_id": "activity_session_history",
                    "read_kind": "lookup",
                    "subject": "session_detail",
                    "filters": { "session_id": "activity_session_019d77b001c472c1ae842d2dba567099" },
                    "limit": 8,
                    "output_shape": "item"
                }
            },
            {
                "label": "bounded block-history aggregation",
                "payload": {
                    "source_id": "activity_session_history",
                    "read_kind": "aggregate",
                    "subject": "block_history",
                    "filters": { "date_from": "2026-04-01" },
                    "query": {
                        "group_by": "comparison_stream",
                        "aggregate": "load_kg_max",
                        "sort": { "field": "load_kg_max", "direction": "desc" }
                    },
                    "limit": 20,
                    "output_shape": "aggregate"
                }
            },
            {
                "label": "list activity blocks carrying a saved semantic metadata tag",
                "payload": {
                    "source_id": "activity_session_history",
                    "read_kind": "list",
                    "subject": "block_history",
                    "filters": { "capability_target": "mobility" },
                    "query": {
                        "fields": ["activity_date", "display_label", "exercise_identity", "block_form", "capability_targets", "body_focus", "execution_traits"],
                        "sort": { "field": "activity_date", "direction": "desc" }
                    },
                    "limit": 50,
                    "output_shape": "collection"
                }
            },
            {
                "label": "CMJ benchmark timeline",
                "payload": {
                    "source_id": "performance_tests",
                    "read_kind": "timeline",
                    "subject": "test_timeline",
                    "filters": {
                        "test_type": "cmj",
                        "measured_property": "jump_height_cm",
                        "date_from": "2026-01-01"
                    },
                    "output_shape": "series"
                }
            },
            {
                "label": "today's planned schedule as a small day record",
                "payload": {
                    "source_id": "training_schedule",
                    "read_kind": "lookup",
                    "subject": "date_bound_schedule",
                    "filters": { "date": "2026-05-04" },
                    "output_shape": "item"
                }
            },
            {
                "label": "paged planned schedule occurrences for a week",
                "payload": {
                    "source_id": "training_schedule",
                    "read_kind": "timeline",
                    "subject": "date_bound_schedule",
                    "filters": { "date_from": "2026-05-04", "date_to": "2026-05-10" },
                    "limit": 5,
                    "output_shape": "series"
                }
            },
            {
                "label": "wearable steps timeline",
                "payload": {
                    "source_id": "wearable_daily",
                    "read_kind": "timeline",
                    "subject": "metric_timeline",
                    "filters": {
                        "metric_key": "steps",
                        "date_from": "2026-01-01"
                    },
                    "output_shape": "series"
                }
            },
            {
                "label": "analyze vocabulary: list observed signals, optionally filtered by domain",
                "payload": {
                    "source_id": "analyze_catalog",
                    "read_kind": "list",
                    "subject": "signals",
                    "filters": { "domain": "recovery" },
                    "output_shape": "collection",
                    "analysis_mode": "none"
                }
            },
            {
                "label": "analyze vocabulary: list observed entities (e.g. exercises the user has logged)",
                "payload": {
                    "source_id": "analyze_catalog",
                    "read_kind": "list",
                    "subject": "entities",
                    "output_shape": "collection",
                    "analysis_mode": "none"
                }
            },
            {
                "label": "analyze vocabulary: page observed analysis candidates with variable-template fills",
                "payload": {
                    "source_id": "analyze_catalog",
                    "read_kind": "list",
                    "subject": "candidates",
                    "filters": { "candidate_type": "group", "domain": "activity" },
                    "limit": 25,
                    "output_shape": "collection",
                    "analysis_mode": "none"
                }
            },
            {
                "label": "analyze vocabulary: search compact candidates by user text",
                "payload": {
                    "source_id": "analyze_catalog",
                    "read_kind": "list",
                    "subject": "vocabulary_search",
                    "filters": { "q": "Front Squat Back Squat", "candidate_type": "entity" },
                    "limit": 5,
                    "output_shape": "collection",
                    "analysis_mode": "none"
                }
            },
            {
                "label": "analyze vocabulary: look up a single signal's observation envelope when analyze asks for catalog lookup",
                "payload": {
                    "source_id": "analyze_catalog",
                    "read_kind": "lookup",
                    "subject": "signal_detail",
                    "filters": { "signal_id": "sleep_quality" },
                    "output_shape": "item",
                    "analysis_mode": "none"
                }
            }
        ],
        "properties": {
            "schema_version": {
                "type": "string",
                "description": "Optional contract version. Requests default to read_query_request.v1."
            },
            "source_id": {
                "type": "string",
                "enum": [
                    "activity_session_history",
                    "activity_protocol_profiles",
                    "training_plan",
                    "training_schedule",
                    "recovery",
                    "performance_tests",
                    "measurements",
                    "wearable_daily",
                    "evidence_fact_health_log",
                    "nutrition",
                    "supplements",
                    "profile",
                    "analyze_catalog"
                ]
            },
            "read_kind": {
                "type": "string",
                "enum": ["lookup", "list", "timeline", "aggregate"]
            },
            "subject": {
                "type": "string",
                "description": "Optional source-specific subject hint. activity_session_history accepts day_history or session_history for compact recall, session_detail or latest_session for exact session detail, block_history for flat programmable activity rows, activity_timeline for ordered day points, or activity_summary for compact totals. activity_protocol_profiles accepts profile_registry, latest_profile, or profile_summary. training_schedule accepts date_bound_schedule/current_day_occurrences/scheduled_session; use read_kind=lookup with filters.date for today's planned schedule, or timeline/list with date_from/date_to plus cursor for a bounded schedule window. performance_tests accepts test_history, latest_test, test_timeline, or test_summary. measurements accepts snapshot_history, latest_snapshot, metric_timeline, or measurement_summary. wearable_daily accepts daily_history, latest_day, metric_timeline, or wearable_summary. profile accepts current_profile, revision_history, profile_timeline, or profile_summary. analyze_catalog accepts candidates (read_kind=list) for the paged observed candidate catalog with handles and analysis_variable_template fills, signals (read_kind=list) for observed signal fields, entities (read_kind=list) for observed entities, vocabulary_search (read_kind=list) for compact ranked candidates by filters.q, or signal_detail (read_kind=lookup) for a single signal envelope. Other readable sources use source-specific subjects and fall back to their default subject when omitted."
            },
            "filters": {
                "type": "object",
                "description": "Optional source-specific filters. activity_session_history supports session_id, source_kind, date, date_from, date_to, session_status, block_kind, analysis_grade, exercise_identity, comparison_stream, and saved block_semantics metadata filters block_form, capability_target (or capability_targets), body_focus, execution_trait (or execution_traits), action_pattern (or action_patterns), and metabolic_profile. Metadata array filters match blocks containing that tag. activity_protocol_profiles supports profile_id and block_kind. training_schedule supports date for one day and date_from/date_to for bounded occurrence windows; use cursor when continuation.has_more=true. performance_tests supports entry_id, date, date_from, date_to, status, test_type, protocol_variant, measured_property, and session_id. measurements supports entry_id, date, date_from, date_to, status, metric_key, metric_family, and source_kind. wearable_daily supports entry_id, date, date_from, date_to, status, metric_key, unit, aggregation, source_kind, and provider. profile supports entry_id, revision_id, date, date_from, date_to, status, write_kind, context_category, and source_kind. analyze_catalog supports domain on candidates/signals/entities/signal_detail, supports signal_id on candidates/entities to list items with observations for a signal, requires signal_id on subject=signal_detail, and requires q on subject=vocabulary_search. candidates and vocabulary_search accept candidate_type=signal|source_signal|scoped_signal|entity|comparison_stream|protocol_variant|body_area|supplement|custom_dimension|candidate_space|group|variable_template. filters.signal is accepted as an alias for signal_id on analyze_catalog. Other readable projection sources use filters.lookup_key only for exact lookup/list item selection on the shared public path.",
                "properties": {
                    "lookup_key": { "type": "string" },
                    "key": { "type": "string" },
                    "profile_id": { "type": "string" },
                    "entry_id": { "type": "string" },
                    "session_id": { "type": "string" },
                    "domain": { "type": "string" },
                    "q": { "type": "string" },
                    "query": { "type": "string" },
                    "candidate_type": {
                        "type": "string",
                        "enum": [
                            "signal",
                            "source_signal",
                            "scoped_signal",
                            "entity",
                            "comparison_stream",
                            "protocol_variant",
                            "body_area",
                            "supplement",
                            "custom_dimension",
                            "candidate_space",
                            "group",
                            "variable_template"
                        ]
                    },
                    "signal_id": { "type": "string" },
                    "signal": { "type": "string" },
                    "source_kind": {
                        "type": "string",
                        "enum": ["manual", "imported", "system", "test_embedded"]
                    },
                    "session_status": {
                        "type": "string",
                        "enum": ["open", "closed", "pending_assignment"]
                    },
                    "block_kind": {
                        "type": "string",
                        "enum": [
                            "repetition_sets",
                            "holds",
                            "timed_intervals",
                            "distance_intervals",
                            "continuous_endurance",
                            "drill",
                            "composite_work",
                            "benchmark_result"
                        ]
                    },
                    "analysis_grade": {
                        "type": "string",
                        "enum": ["comparable", "limited", "context_only"]
                    },
                    "exercise_identity": { "type": "string" },
                    "comparison_stream": { "type": "string" },
                    "block_form": { "type": "string" },
                    "capability_target": { "type": "string" },
                    "capability_targets": { "type": "string" },
                    "body_focus": { "type": "string" },
                    "execution_trait": { "type": "string" },
                    "execution_traits": { "type": "string" },
                    "action_pattern": { "type": "string" },
                    "action_patterns": { "type": "string" },
                    "metabolic_profile": { "type": "string" },
                    "status": {
                        "type": "string",
                        "enum": ["active", "retracted"]
                    },
                    "test_type": { "type": "string" },
                    "protocol_variant": { "type": "string" },
                    "measured_property": { "type": "string" },
                    "metric_key": { "type": "string" },
                    "metric_family": { "type": "string" },
                    "unit": { "type": "string" },
                    "aggregation": { "type": "string" },
                    "provider": { "type": "string" },
                    "date": { "type": "string" },
                    "date_from": { "type": "string" },
                    "date_to": { "type": "string" }
                },
                "additionalProperties": true
            },
            "query": {
                "type": "object",
                "description": "Optional bounded programmable read program. activity_session_history supports this only on compact subjects session_history and block_history. For subject=block_history, query.fields can include saved semantic metadata fields block_form, capability_targets, body_focus, execution_traits, action_patterns, metabolic_profile, or block_semantics. Use read_kind=list with query.fields and optional query.sort for compact row retrieval, or read_kind=aggregate with query.aggregate plus optional query.group_by/query.sort for grouped summaries. Metadata array group_by fields create one group per tag value.",
                "properties": {
                    "fields": {
                        "type": "array",
                        "items": { "type": "string" }
                    },
                    "group_by": { "type": "string" },
                    "aggregate": { "type": "string" },
                    "sort": {
                        "type": "object",
                        "properties": {
                            "field": { "type": "string" },
                            "direction": {
                                "type": "string",
                                "enum": ["asc", "desc"]
                            }
                        },
                        "required": ["field"],
                        "additionalProperties": false
                    }
                },
                "additionalProperties": false
            },
            "limit": {
                "type": "integer",
                "minimum": 1,
                "maximum": 100,
                "description": "Optional page size for exact result pages. The backend validates source-specific continuation behavior."
            },
            "cursor": {
                "type": "string",
                "description": "Opaque server-issued continuation cursor from the previous read_query response for the same source and query semantics."
            },
            "output_shape": {
                "type": "string",
                "enum": ["item", "collection", "series", "aggregate"]
            },
            "analysis_mode": {
                "type": "string",
                "enum": ["none", "allow_hybrid", "require_analysis"]
            }
        },
        "required": ["source_id", "read_kind"],
        "additionalProperties": false
    })
}

fn activity_block_kind_schema() -> Value {
    json!({
        "type": "string",
        "enum": [
            "repetition_sets",
            "holds",
            "timed_intervals",
            "distance_intervals",
            "continuous_endurance",
            "drill",
            "composite_work",
            "benchmark_result"
        ]
    })
}

fn activity_block_semantics_schema() -> Value {
    json!({
        "type": "object",
        "description": "Saved primitive block semantics. Use this instead of analysis_family_tags. Keep it small and literal: one block_form plus bounded arrays for capability_targets, action_patterns, metabolic_profile, body_focus, and execution_traits.",
        "properties": {
            "schema_version": {
                "type": "string",
                "enum": ["activity_block_semantics.v1"]
            },
            "block_form": {
                "type": "string",
                "enum": [
                    "exercise",
                    "drill",
                    "interval",
                    "skill",
                    "sparring",
                    "conditioning",
                    "mobility",
                    "rehab",
                    "warmup",
                    "test",
                    "play"
                ]
            },
            "capability_targets": { "type": "array", "items": { "type": "string" } },
            "action_patterns": { "type": "array", "items": { "type": "string" } },
            "metabolic_profile": { "type": "array", "items": { "type": "string" } },
            "body_focus": { "type": "array", "items": { "type": "string" } },
            "execution_traits": { "type": "array", "items": { "type": "string" } }
        },
        "additionalProperties": true
    })
}

fn record_activity_vnext_schema() -> Value {
    json!({
        "type": "object",
        "description": "One task-first activity write on /v4/agent/record-activity. Use date plus blocks only for executed activity sessions. If the user mentions a future date, use plan_activity unless the user explicitly says it already happened; then send occurrence_status=executed. Stay on this same surface for saved block corrections with operation=remove|replace plus target and replacement_patch, for session follow-up with target_session_id plus close_session/session_follow_up, for stable visible exercise repairs with display_label_action.action=rename or display_label_action.action=clear, for exercise-identity merge previews with exercise_identity_merge.action=preview, and for one bounded blocked-write continuation when the server returns needs_input with continuation. When you recognize a stable exercise identity and the visible exercise name is clear, set display_label on the first clean log instead of waiting for a later repair. Successful writes may return active follow_ups for analysis gaps, session closeout, comparability context, new comparison streams inside an exercise family, risky exercise identity reuse, stable display-label completion, or semantic completion for block_semantics. Ask user-facing follow_ups verbatim when requested, but display-label and semantic completion tasks are agent-side active work cards: process every returned target, first reuse one exact known template when available, otherwise fill the bounded display_label or block_semantics yourself and ask the user only if the visible name or exercise meaning is genuinely unclear. Use display_label only for the stable canonical visible exercise name you want read and analyze to reuse later; keep label as the raw block label for this saved block. display_label_action repairs the stable visible name only; it must not change exercise identities. exercise_identity_merge only prepares a merge preview; the actual merge runs only after an explicit user-approved continuation. Only send protocol_profile when the user has confirmed a recurring comparable protocol; the server does not infer or create protocol profiles from free text. If the user confirms a profile for one already saved block, send one atomic operation=replace payload with target.item_id, replacement_patch, and protocol_profile.apply_to_block_id matching that same block.",
        "properties": {
            "continuation": {
                "type": "object",
                "description": "Server-owned bounded continuation for a blocked record_activity write. Use only after record_activity returned needs_input with a continuation object, and send only the exact continuation object with one filled answer field. If the workout content changed, start a fresh record_activity write instead.",
                "properties": {
                    "mode": { "type": "string" },
                    "clarification_id": { "type": "string" },
                    "selected_exercise_identity": { "type": "string" },
                    "confirm_new_exercise": { "type": "boolean" },
                    "approve_merge": { "type": "boolean" }
                },
                "required": ["clarification_id"],
                "additionalProperties": false
            },
            "date": { "type": "string" },
            "occurrence_status": {
                "type": "string",
                "enum": ["executed", "planned"],
                "description": "Use occurrence_status=executed only when a future-dated activity already happened. Planned future training belongs on plan_activity, not record_activity."
            },
            "blocks": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "kind": activity_block_kind_schema(),
                        "label": { "type": "string" },
                        "display_label": { "type": "string" },
                        "measurements": { "type": "object", "additionalProperties": true },
                        "exercise_identity": { "type": "string" },
                        "intent": { "type": "string" },
                        "modality": { "type": "string" },
                        "protocol_profile_id": { "type": "string" },
                        "execution_context": { "type": "object", "additionalProperties": true },
                        "block_semantics": activity_block_semantics_schema(),
                        "analysis_grade": {
                            "type": "string",
                            "enum": ["comparable", "limited", "context_only"]
                        }
                    },
                    "required": ["kind", "measurements"],
                    "additionalProperties": false
                }
            },
            "operation": { "type": "string", "enum": ["remove", "replace"] },
            "target": {
                "type": "object",
                "properties": {
                    "session_id": { "type": "string" },
                    "item_id": { "type": "string" }
                },
                "additionalProperties": true
            },
            "replacement_patch": { "type": "object", "additionalProperties": true },
            "display_label_action": {
                "type": "object",
                "description": "Stable visible exercise repair for already saved activity data. Use action=rename to set the canonical visible display name or action=clear to remove a wrong stable visible display name without changing exercise_identity.",
                "properties": {
                    "action": { "type": "string", "enum": ["rename", "clear"] },
                    "display_label": { "type": "string" },
                    "exercise_identity": { "type": "string" },
                    "comparison_stream_key": { "type": "string" },
                    "profile_id": { "type": "string" },
                    "reason": { "type": "string" }
                },
                "required": ["action"],
                "additionalProperties": false
            },
            "exercise_identity_merge": {
                "type": "object",
                "description": "User-scoped exercise identity merge preview. This only prepares a preview and approval continuation; the merge itself runs only after explicit user approval.",
                "properties": {
                    "action": { "type": "string", "enum": ["preview"] },
                    "surviving_exercise_identity": { "type": "string" },
                    "source_exercise_identities": {
                        "type": "array",
                        "items": { "type": "string" }
                    },
                    "reason": { "type": "string" }
                },
                "required": ["action", "surviving_exercise_identity", "source_exercise_identities"],
                "additionalProperties": false
            },
            "target_session_id": { "type": "string" },
            "start_new_session": { "type": "boolean" },
            "close_session": { "type": "boolean" },
            "session_follow_up": { "type": "object", "additionalProperties": true },
            "protocol_profile": {
                "type": "object",
                "description": "Explicit agent-selected protocol-profile action. Use after the user confirms that a setup is a recurring comparable protocol, for example Flying 30m with the same run-in. Do not use this for ordinary repeated exercises or one-off context.",
                "properties": {
                    "action": { "type": "string", "enum": ["create", "revise"] },
                    "apply_to_block_index": { "type": "integer", "minimum": 0 },
                    "apply_to_block_id": {
                        "type": "string",
                        "description": "Use only with a saved-block profile confirmation: operation=replace plus target.item_id for the same saved block. For fresh blocks submitted in this same call, use apply_to_block_index instead."
                    },
                    "schema_version": { "type": "string" },
                    "profile_id": { "type": "string" },
                    "expected_previous_revision": { "type": "integer" },
                    "profile_label": { "type": "string" },
                    "block_kind": activity_block_kind_schema(),
                    "defaults": {
                        "type": "object",
                        "description": "Bounded defaults for this recurring protocol, such as primary_label, display_label, exercise_identity, measurement_defaults, execution_context_defaults, intent, modality, or block_semantics.",
                        "additionalProperties": true
                    },
                    "comparability": {
                        "type": "object",
                        "description": "Fields required for later comparable analysis, especially required_measurement_keys and required_execution_context_keys.",
                        "additionalProperties": true
                    },
                    "metadata": { "type": "object", "additionalProperties": true },
                    "idempotency_key": { "type": "string" }
                },
                "required": ["action", "profile_label", "block_kind"],
                "additionalProperties": false
            },
            "notes": { "type": "string" },
            "session_context": { "type": "object", "additionalProperties": true },
            "idempotency_key": { "type": "string" }
        },
        "additionalProperties": false
    })
}

fn record_recovery_vnext_schema() -> Value {
    let mut schema = write_recovery_vnext_schema();
    if let Some(description) = schema.get_mut("description") {
        *description = json!(
            "One task-first recovery write on /v4/agent/record-recovery. bodyweight_kg may accompany a real recovery check-in only as a measurement handoff: Kura stores recovery signals in recovery and stores bodyweight in measurements, not as a recovery signal. Explicit same-day check-ins are separate observations; use timestamp for morning/evening or historical check-ins. For later same-day check-ins, send only fields the user explicitly mentioned; do not copy sleep_hours or bodyweight_kg from an earlier check-in unless the user confirmed a nap/new sleep entry or a new weigh-in. body_symptoms are canonical recurring metrics: if Kura returns needs_input for an unknown body symptom metric, resolve it on this same tool with only the returned continuation object. If Kura flags a duplicate same-day value and the user confirms it, retry with same_day_confirmed_signals such as [\"sleep_hours\"]. To fix a saved recovery check-in, stay on record_recovery and send operation=remove|replace; measurement corrections stay on record_measurement."
        );
    }
    schema
}

fn report_health_vnext_schema() -> Value {
    let mut schema = write_health_vnext_schema();
    if let Some(description) = schema.get_mut("description") {
        *description = json!(
            "One task-first health write on /v4/agent/report-health. Use kind=injury or kind=illness with canonical health fields only. To fix one active health state, stay on report_health with operation=update|resolve|remove_mistaken_report."
        );
    }
    schema
}

fn record_nutrition_vnext_schema() -> Value {
    let mut schema = write_nutrition_vnext_schema();
    if let Some(description) = schema.get_mut("description") {
        *description = json!(
            "One task-first nutrition write on /v4/agent/record-nutrition. Use kind=meal for meal logs, including photo/text-derived estimates with evidence_source and confidence fields, or kind=target for nutrition targets. Saved meal corrections stay on record_nutrition with operation=remove|replace."
        );
    }
    schema
}

fn record_measurement_vnext_schema() -> Value {
    let mut schema = write_measurement_vnext_schema();
    if let Some(description) = schema.get_mut("description") {
        *description = json!(
            "One task-first measurement write on /v4/agent/record-measurement. Use kind=bodyweight for bodyweight/body-composition fields or kind=measurement for anthropometric metrics. Saved measurement corrections stay on record_measurement with operation=remove|replace."
        );
    }
    schema
}

fn record_benchmark_vnext_schema() -> Value {
    let mut schema = write_performance_test_vnext_schema();
    if let Some(description) = schema.get_mut("description") {
        *description = json!(
            "One task-first benchmark write on /v4/agent/record-benchmark. Use canonical performance_test fields for explicit benchmark or assessment results only, not for normal training volume. Saved benchmark corrections stay on record_benchmark with operation=remove|replace."
        );
    }
    schema
}

fn update_profile_vnext_schema() -> Value {
    json!({
        "type": "object",
        "description": "One task-first profile update on /v4/agent/update-profile. Use this for durable profile context plus explicitly allowed measurement and baseline-benchmark fanout when the user provides them together.",
        "properties": {
            "preferred_name": { "type": "string" },
            "experience_level": { "type": "string" },
            "training_modality": { "type": "string" },
            "training_frequency_per_week": { "type": "number" },
            "available_equipment": { "type": "array", "items": { "type": "string" } },
            "primary_location": { "type": "string" },
            "communication_style": { "type": "string" },
            "age": { "type": "number" },
            "notes": { "type": "string" },
            "training_background": { "type": "object", "additionalProperties": true },
            "goal": { "type": "object", "additionalProperties": true },
            "goals": { "type": "array", "items": { "type": "object", "additionalProperties": true } },
            "preference": { "type": "object", "additionalProperties": true },
            "preferences": { "type": "array", "items": { "type": "object", "additionalProperties": true } },
            "constraints": { "type": "object", "additionalProperties": true },
            "stature_cm": { "type": "number" },
            "wingspan_cm": { "type": "number" },
            "bodyweight_kg": { "type": "number" },
            "baseline_benchmark": {
                "type": "object",
                "properties": {
                    "test_type": { "type": "string" },
                    "protocol_variant": { "type": "string" },
                    "result_value": { "type": "number" },
                    "result_unit": { "type": "string" }
                },
                "additionalProperties": true
            },
            "idempotency_key": { "type": "string" }
        },
        "additionalProperties": false
    })
}

fn plan_activity_vnext_schema() -> Value {
    json!({
        "type": "object",
        "description": "One task-first planned activity write on /v4/agent/plan-activity. Use this when the user describes future or not-yet-executed training such as 'next week'. Do not use record_activity unless the user says it already happened. Send date plus training-shaped blocks[] or at least a session_name placeholder.",
        "properties": {
            "date": {
                "type": "string",
                "description": "Absolute planned local date in YYYY-MM-DD."
            },
            "local_date": {
                "type": "string",
                "description": "Alias for date. If both date and local_date are present, they must match."
            },
            "exception_id": { "type": "string" },
            "session_name": { "type": "string" },
            "name": { "type": "string" },
            "session_key": { "type": "string" },
            "blocks": {
                "type": "array",
                "description": "Training-shaped planned blocks. Use the same block language as record_activity when the user gives exact exercises, sets, reps, loads, and variant context.",
                "items": {
                    "type": "object",
                    "properties": {
                        "kind": activity_block_kind_schema(),
                        "label": { "type": "string" },
                        "exercise_identity": { "type": "string" },
                        "measurements": { "type": "object", "additionalProperties": true },
                        "execution_context": { "type": "object", "additionalProperties": true },
                        "intent": { "type": "string" },
                        "modality": { "type": "string" },
                        "protocol_profile_id": { "type": "string" }
                    },
                    "additionalProperties": true
                }
            },
            "exercises": {
                "type": "array",
                "description": "Optional simpler planned exercise list when blocks[] is not available.",
                "items": { "type": "object", "additionalProperties": true }
            },
            "location": { "type": "string" },
            "start_time": { "type": "string" },
            "end_time": { "type": "string" },
            "notes": { "type": "string" },
            "note": { "type": "string" },
            "reason": { "type": "string" },
            "timezone": { "type": "string" },
            "session_context": { "type": "object", "additionalProperties": true },
            "progression_override": { "type": "object", "additionalProperties": true },
            "timestamp": { "type": "string" },
            "idempotency_key": { "type": "string" }
        },
        "required": ["date"],
        "additionalProperties": false
    })
}

fn update_plan_vnext_schema() -> Value {
    let mut schema = write_plan_vnext_schema();
    if let Some(description) = schema.get_mut("description") {
        *description = json!(
            "One task-first planning write on /v4/agent/update-plan. Use kind=create, update, replace, archive, schedule_exception_upsert, or schedule_exception_clear for normal plan and schedule changes."
        );
    }
    schema
}

fn write_recovery_vnext_schema() -> Value {
    json!({
        "type": "object",
        "description": "One canonical recovery check-in observation. bodyweight_kg may accompany a real recovery check-in only as a measurement handoff: Kura stores recovery signals in recovery and stores bodyweight in measurements, not as a recovery signal. Use timestamp on this same surface for multiple same-day check-ins or historical recovery check-ins such as yesterday morning or a prior evening check-in. Same-day follow-up check-ins may be partial: send only fields the user explicitly mentioned and do not copy prior sleep or weight values unless the user confirms a nap/new sleep entry or a new weigh-in. If Kura flags a duplicate same-day value and the user confirms it, retry with same_day_confirmed_signals such as [\"sleep_hours\"]. To fix a saved recovery check-in, stay on write_recovery and send operation=remove|replace. Measurement corrections stay on write_measurement/record_measurement. Omit target.event_id only when the saved check-in is uniquely determined; send target.event_id to pin one exact saved entry or when Kura asks you to choose.",
        "properties": {
            "operation": { "type": "string", "enum": ["remove", "replace"] },
            "continuation": {
                "type": "object",
                "description": "Server-owned bounded continuation for a blocked record_recovery body_symptoms metric. Use only after record_recovery returned needs_input with continuation; send the exact continuation object with exactly one of selected_metric_id or confirm_new_metric=true.",
                "properties": {
                    "clarification_id": { "type": "string" },
                    "mode": { "type": "string" },
                    "selected_metric_id": { "type": "string" },
                    "confirm_new_metric": { "type": "boolean" },
                    "answer_template": { "type": "object", "additionalProperties": true }
                },
                "required": ["clarification_id"],
                "additionalProperties": true
            },
            "target": {
                "type": "object",
                "properties": {
                    "event_id": { "type": "string" }
                },
                "additionalProperties": false
            },
            "sleep_hours": { "type": "number" },
            "energy_level": { "type": "number" },
            "soreness": { "type": "number" },
            "sleep_quality": { "type": "number" },
            "sleep_score": { "type": "number" },
            "quality": { "type": "number" },
            "readiness_score": { "type": "number" },
            "bed_time": { "type": "string" },
            "wake_time": { "type": "string" },
            "soreness_areas": { "type": "array", "items": { "type": "string" }, "description": "Affected body areas when the user names them, for example [\"legs\", \"lower_back\"]." },
            "body_symptoms": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "body_area": { "type": "string" },
                        "signal": { "type": "string" },
                        "severity": { "type": "number" }
                    },
                    "required": ["body_area", "signal", "severity"],
                    "additionalProperties": false
                },
                "description": "Daily body symptom entries for structured pain/stiffness tracking, for example [{\"body_area\":\"right_knee\",\"signal\":\"pain\",\"severity\":5}]. Existing user metrics are reused by exact canonical body_area + signal. Unknown metrics are blocked with candidates/confirmation instead of being silently created."
            },
            "notes": { "type": "string" },
            "bodyweight_kg": { "type": "number", "description": "Measurement handoff only. Kura stores this in measurements, not as a recovery signal." },
            "motivation": { "type": "number" },
            "physical_condition": { "type": "string" },
            "lifestyle_stability": { "type": "string" },
            "alcohol_last_night": { "type": "boolean" },
            "sick_today": { "type": "boolean" },
            "training_yesterday": { "type": "boolean" },
            "traveling_yesterday": { "type": "boolean" },
            "hrv_rmssd": { "type": "number" },
            "supplements": { "type": "array", "items": { "type": "string" } },
            "compact_input": { "type": "string" },
            "same_day_confirmed_signals": {
                "type": "array",
                "items": { "type": "string" },
                "description": "Use only after Kura flags a duplicate same-day signal and the user confirms the new check-in really repeated that value, for example [\"sleep_hours\"]."
            },
            "timestamp": { "type": "string", "description": "RFC3339 timestamp. Use this to backdate historical recovery check-ins on the same surface." },
            "idempotency_key": { "type": "string" }
        },
        "additionalProperties": false
    })
}

fn write_health_vnext_schema() -> Value {
    json!({
        "type": "object",
        "description": "One health-state write. Use kind=injury or kind=illness with canonical health fields only. Use timestamp on this same surface to backdate historical injury or illness reports instead of routing them elsewhere. To fix one active health state, stay on write_health with operation=update|resolve|remove_mistaken_report. Omit target.event_id when the active state is uniquely determined; send target.event_id only to pin one exact report or when Kura asks you to choose.",
        "properties": {
            "operation": {
                "type": "string",
                "enum": ["update", "resolve", "remove_mistaken_report"]
            },
            "target": {
                "type": "object",
                "properties": {
                    "event_id": { "type": "string" }
                },
                "additionalProperties": false
            },
            "kind": { "type": "string", "enum": ["injury", "illness"] },
            "affected_area": { "type": "string" },
            "description": { "type": "string" },
            "severity": { "type": "number" },
            "since": { "type": "string" },
            "label": { "type": "string" },
            "status": { "type": "string" },
            "timestamp": { "type": "string", "description": "RFC3339 timestamp. Use this to backdate historical injury or illness reports on the same surface." },
            "idempotency_key": { "type": "string" }
        },
        "required": ["kind"],
        "additionalProperties": false
    })
}

fn write_nutrition_vnext_schema() -> Value {
    json!({
        "type": "object",
        "description": "One nutrition-family write. Use kind=meal for meal logs or kind=target for nutrition targets. For photo/text-derived meals, save macro estimates together with evidence_source, estimate_confidence, portion_confidence, assumptions, items, and needs_confirmation when relevant so reads and analysis do not treat estimates as exact weighed data. Use timestamp on this same surface to backdate historical meals or targets instead of hiding meal timing in notes. Saved meal corrections stay on write_nutrition with operation=remove|replace; omit target.event_id when the saved meal is uniquely determined and send target.event_id only for exact or ambiguous follow-up choices.",
        "properties": {
            "operation": { "type": "string", "enum": ["remove", "replace"] },
            "target": {
                "type": "object",
                "properties": {
                    "event_id": { "type": "string" }
                },
                "additionalProperties": false
            },
            "kind": { "type": "string", "enum": ["meal", "target"] },
            "description": { "type": "string" },
            "meal_type": { "type": "string" },
            "entry_scope": {
                "type": "string",
                "description": "Scope of the nutrition entry, usually meal; use day_total only when the user gives a whole-day total."
            },
            "evidence_source": {
                "type": "string",
                "description": "How the meal estimate was derived, for example manual, text, image, nutrition_label, restaurant_menu, or import."
            },
            "estimate_confidence": {
                "type": "number",
                "description": "Confidence in the macro estimate, 0..1."
            },
            "portion_confidence": {
                "type": "number",
                "description": "Confidence in portion-size recognition, 0..1."
            },
            "needs_confirmation": {
                "type": "boolean",
                "description": "True when the user should verify the estimated meal."
            },
            "assumptions": {
                "type": "array",
                "description": "Assumptions made while estimating the meal.",
                "items": { "type": "string" }
            },
            "items": {
                "type": "array",
                "description": "Recognized foods with optional quantities and macro estimates.",
                "items": { "type": "object", "additionalProperties": true }
            },
            "evidence_ref": {
                "type": "string",
                "description": "Optional opaque reference for the image, import, or label source."
            },
            "evidence_notes": {
                "type": "string",
                "description": "Optional notes about evidence quality or estimation caveats."
            },
            "calories": { "type": "number" },
            "protein_g": { "type": "number" },
            "carbs_g": { "type": "number" },
            "fat_g": { "type": "number" },
            "fiber_g": { "type": "number" },
            "added_sugar_g": { "type": "number" },
            "saturated_fat_g": { "type": "number" },
            "sodium_mg": { "type": "number" },
            "alcohol_units": { "type": "number" },
            "target_calories": { "type": "number" },
            "target_protein_g": { "type": "number" },
            "target_carbs_g": { "type": "number" },
            "target_fat_g": { "type": "number" },
            "target_fiber_g": { "type": "number" },
            "target_added_sugar_g_max": { "type": "number" },
            "target_saturated_fat_g_max": { "type": "number" },
            "target_sodium_mg_max": { "type": "number" },
            "target_alcohol_units_max": { "type": "number" },
            "timestamp": { "type": "string", "description": "RFC3339 timestamp. Use this to backdate historical meals or nutrition targets on the same surface." },
            "idempotency_key": { "type": "string" }
        },
        "required": ["kind"],
        "additionalProperties": false
    })
}

fn write_performance_test_vnext_schema() -> Value {
    json!({
        "type": "object",
        "description": "One explicit benchmark or assessment result on /v4/agent/write-event. Use canonical performance_test.logged fields only. This surface is for formal comparable tests such as CMJ, rim touch, broad jump, sprint timing, or similar benchmark artifacts. To fix one saved benchmark, stay on write_performance_test with operation=remove|replace. remove still needs target.event_id; replace may omit target.event_id when one saved test is uniquely determined.",
        "properties": {
            "operation": { "type": "string", "enum": ["remove", "replace"] },
            "target": {
                "type": "object",
                "properties": {
                    "event_id": { "type": "string" }
                },
                "additionalProperties": false
            },
            "test_type": { "type": "string" },
            "protocol_variant": { "type": "string" },
            "measured_property": { "type": "string" },
            "result": {
                "type": "object",
                "properties": {
                    "kind": { "type": "string", "enum": ["numeric", "boolean", "text", "ordinal"] },
                    "value": {},
                    "unit": { "type": "string" },
                    "attempts": {
                        "type": "array",
                        "description": "Individual trial values when the user gives multiple attempts, e.g. [42, 41, 43] for a CMJ series. Keep result.value as the primary/best reported value.",
                        "items": {}
                    }
                },
                "required": ["value"],
                "additionalProperties": false
            },
            "attempts": {
                "type": "array",
                "description": "Optional top-level alias for result.attempts; use result.attempts when possible.",
                "items": {}
            },
            "contact_level": { "type": "string" },
            "auxiliary_metrics": {
                "type": "array",
                "items": { "type": "object", "additionalProperties": true }
            },
            "context_text": { "type": "string" },
            "device_type": { "type": "string" },
            "surface": { "type": "string" },
            "session_id": { "type": "string" },
            "timestamp": { "type": "string", "description": "RFC3339 timestamp. Use this to backdate historical benchmark entries on the same surface." },
            "idempotency_key": { "type": "string" }
        },
        "additionalProperties": false
    })
}

fn write_measurement_vnext_schema() -> Value {
    json!({
        "type": "object",
        "description": "One measurements-family write. Use kind=bodyweight for bodyweight/body-composition fields or kind=measurement for circumference-style metrics. Use timestamp on this same surface to backdate historical weigh-ins or measurement checks instead of hiding timing in notes. Saved measurement corrections stay on write_measurement with operation=remove|replace; omit target.event_id when the saved entry is uniquely determined and send target.event_id only for exact or ambiguous follow-up choices.",
        "properties": {
            "operation": { "type": "string", "enum": ["remove", "replace"] },
            "target": {
                "type": "object",
                "properties": {
                    "event_id": { "type": "string" }
                },
                "additionalProperties": false
            },
            "kind": { "type": "string", "enum": ["bodyweight", "measurement"] },
            "weight_kg": { "type": "number" },
            "body_fat_pct": { "type": "number" },
            "skeletal_muscle_mass_lbs": { "type": "number" },
            "time_of_day": { "type": "string" },
            "conditions": { "type": "string" },
            "type": { "type": "string" },
            "value_cm": { "type": "number" },
            "side": { "type": "string" },
            "timestamp": { "type": "string", "description": "RFC3339 timestamp. Use this to backdate historical bodyweight or measurement entries on the same surface." },
            "idempotency_key": { "type": "string" }
        },
        "required": ["kind"],
        "additionalProperties": false
    })
}

fn write_plan_vnext_schema() -> Value {
    json!({
        "type": "object",
        "description": "One planning-family write on /v4/agent/write-event. Use exactly one primary payload: either a fresh plan/schedule write with kind=create|update|replace|archive|schedule_exception_upsert|schedule_exception_clear, or a pending_action_confirmation retry when Kura returned a pending_action_confirmation_challenge. Fresh writes may include read_after_write_targets for verification. When pending_action_confirmation is present, send only that field instead of replaying the original structured plan payload.",
        "properties": {
            "kind": {
                "type": "string",
                "enum": [
                    "create",
                    "update",
                    "replace",
                    "archive",
                    "schedule_exception_upsert",
                    "schedule_exception_clear"
                ]
            },
            "read_after_write_targets": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "projection_type": { "type": "string" },
                        "key": { "type": "string" }
                    },
                    "required": ["projection_type", "key"],
                    "additionalProperties": false
                }
            },
            "pending_action_confirmation": {
                "type": "object",
                "properties": {
                    "schema_version": { "type": "string" },
                    "confirmed": { "type": "boolean" },
                    "confirmed_at": { "type": "string" },
                    "pending_confirmation_id": { "type": "string" }
                },
                "additionalProperties": false
            },
            "name": { "type": "string" },
            "plan_id": { "type": "string" },
            "sessions": { "type": "array", "items": { "type": "object" } },
            "cycle_weeks": { "type": "number" },
            "start_date": { "type": "string" },
            "phase": { "type": "string" },
            "frequency_per_week": { "type": "number" },
            "sessions_per_week": { "type": "number" },
            "progression": { "type": "object" },
            "notes": { "type": "string" },
            "change_scope": { "type": "string" },
            "update_scope": { "type": "string" },
            "replace_entire_plan": { "type": "boolean" },
            "archive_previous_plan": { "type": "boolean" },
            "requires_explicit_confirmation": { "type": "boolean" },
            "delta": { "type": "object" },
            "reason": { "type": "string" },
            "exception_id": { "type": "string" },
            "operation": { "type": "string" },
            "selector": { "type": "object" },
            "target": { "type": "object" },
            "patch": { "type": "object" },
            "changes": { "type": "object" },
            "progression_override": { "type": "object" },
            "occurrence": { "type": "object" },
            "scope": { "type": "object" },
            "expires_at": { "type": "string" },
            "timestamp": { "type": "string" },
            "idempotency_key": { "type": "string" }
        },
        "required": ["kind"],
        "additionalProperties": false
    })
}

fn vnext_tool_definitions() -> Vec<ToolDefinition> {
    vec![
        ToolDefinition {
            name: "read_query",
            description: "Universal canonical read surface on /v4/agent/read-query. Use this for saved sessions, protocol profiles, recovery, health, measurements, wearable daily metrics, performance tests, profile, nutrition, supplements, planning, exact records, and raw timelines. For today's planned training or a specific planned date, use source_id=training_schedule read_kind=lookup with filters.date; for planned schedule windows, use read_kind=timeline/list with date_from/date_to and continue with the returned cursor when continuation.has_more=true. Use analyze first for trend, comparison, association, explanation, stagnation, driver, influence, cross-domain, or decision questions; do not use read_query as routine analysis preflight. Activity-session summaries expose comparison_stream as the default exact progress grouping and exercise_identity only as a family/mixed-variant view; block_history also exposes saved block_semantics metadata such as block_form, capability_targets, body_focus, execution_traits, action_patterns, and metabolic_profile for bounded fact reads.",
            input_schema: read_query_vnext_schema(),
        },
        ToolDefinition {
            name: "analyze",
            description: analyze_tool_description(),
            input_schema: analyze_input_schema(),
        },
        ToolDefinition {
            name: "record_activity",
            description: "Task-first activity write on /v4/agent/record-activity. Use this as the only normal public activity logging tool. Send date plus blocks[] for fresh writes, include analysis context such as load/result unit and set effort when the user provides it, and set display_label on the first clean log whenever the stable visible name is already clear. Stay on this same tool for saved-block corrections, saved display-label repairs, exercise-identity merge previews, or session follow-up. If status=success includes follow_ups, treat them as active work: ask user-facing questions verbatim, and complete agent-side display-label or block-semantics tasks directly on record_activity for every returned target instead of treating them as passive hints. If the user confirms a recurring comparable protocol, use protocol_profile.action=create|revise explicitly; for one already saved block use operation=replace plus target.item_id, replacement_patch, and protocol_profile.apply_to_block_id in one call. Do not expect the server to infer protocol profiles from ordinary activity text.",
            input_schema: record_activity_vnext_schema(),
        },
        ToolDefinition {
            name: "record_recovery",
            description: "Task-first recovery write on /v4/agent/record-recovery. Use canonical recovery check-in fields only. Multiple same-day check-ins are separate observations when the user gives a new check-in; later same-day check-ins may be partial and must not copy prior sleep or weight values unless the user explicitly confirms them. Corrections stay on the same tool and require the exact ref if same-day entries are ambiguous. bodyweight_kg may accompany a real recovery check-in only as a measurement handoff; Kura stores it in measurements, not recovery.",
            input_schema: record_recovery_vnext_schema(),
        },
        ToolDefinition {
            name: "report_health",
            description: "Task-first health write on /v4/agent/report-health. Use kind=injury or kind=illness with canonical health fields only. To fix one active health state, stay on report_health with operation=update|resolve|remove_mistaken_report.",
            input_schema: report_health_vnext_schema(),
        },
        ToolDefinition {
            name: "record_nutrition",
            description: "Task-first nutrition write on /v4/agent/record-nutrition. Use kind=meal for meal logs or kind=target for nutrition targets with canonical nutrition fields only. Saved meal corrections stay on the same tool with operation=remove|replace.",
            input_schema: record_nutrition_vnext_schema(),
        },
        ToolDefinition {
            name: "record_measurement",
            description: "Task-first measurement write on /v4/agent/record-measurement. Use kind=bodyweight for bodyweight/body-composition fields or kind=measurement for circumference-style metrics. Saved measurement corrections stay on the same tool with operation=remove|replace.",
            input_schema: record_measurement_vnext_schema(),
        },
        ToolDefinition {
            name: "record_benchmark",
            description: "Task-first benchmark write on /v4/agent/record-benchmark. Use canonical performance-test fields for explicit benchmark or assessment results only, not for normal training volume. Saved benchmark corrections stay on the same tool with operation=remove|replace.",
            input_schema: record_benchmark_vnext_schema(),
        },
        ToolDefinition {
            name: "update_profile",
            description: "Task-first profile update on /v4/agent/update-profile. Use this for durable profile context and explicitly allowed measurement or baseline-benchmark fanout when the user gives them together.",
            input_schema: update_profile_vnext_schema(),
        },
        ToolDefinition {
            name: "plan_activity",
            description: "Task-first planned activity write on /v4/agent/plan-activity. Use this for future or not-yet-executed training so planned workouts do not become open activity logs.",
            input_schema: plan_activity_vnext_schema(),
        },
        ToolDefinition {
            name: "update_plan",
            description: "Task-first planning write on /v4/agent/update-plan. Use kind=create, update, replace, archive, schedule_exception_upsert, or schedule_exception_clear for normal plan and schedule changes.",
            input_schema: update_plan_vnext_schema(),
        },
    ]
}

fn resource_definitions() -> Vec<ResourceDefinition> {
    vec![
        ResourceDefinition {
            uri: "kura://openapi",
            name: "Kura OpenAPI Spec",
            description: "Live OpenAPI schema from /api-doc/openapi.json",
        },
        ResourceDefinition {
            uri: "kura://agent/capabilities",
            name: "Agent Capabilities Contract",
            description: "Current write/read protocol expectations for agents",
        },
        ResourceDefinition {
            uri: "kura://agent/skillpacks",
            name: "Agent Skillpack Brief",
            description: "Current global agent skillpack brief for this MCP surface",
        },
        ResourceDefinition {
            uri: "kura://system/config",
            name: "System Config",
            description: "Global dimensions, conventions, and static agent config",
        },
        ResourceDefinition {
            uri: "kura://system/config/manifest",
            name: "System Config Manifest",
            description: "Machine-readable section index with fetch contracts and size hints",
        },
    ]
}

fn append_skillpack_brief_to_initialize_instructions(
    mut instructions: String,
    skillpack_brief: Option<&str>,
) -> String {
    if let Some(brief) = skillpack_brief
        .map(str::trim)
        .filter(|brief| !brief.is_empty())
    {
        instructions.push_str(" Skillpack brief: ");
        instructions.push_str(brief);
    }
    instructions
}

fn tool_completion_status(payload: &Value) -> &'static str {
    if payload
        .get("compatibility")
        .and_then(|v| v.get("fallback_applied"))
        .and_then(Value::as_bool)
        .unwrap_or(false)
    {
        "complete_with_fallback"
    } else {
        "complete"
    }
}

fn extract_auth_me_identity(body: &Value) -> (Option<String>, Option<String>) {
    let user_id = body
        .get("user_id")
        .and_then(Value::as_str)
        .and_then(normalize_user_id_for_allowlist);
    let email = body
        .get("email")
        .and_then(Value::as_str)
        .and_then(normalize_email_for_allowlist);
    (user_id, email)
}

fn tool_text_content(_tool: &str, envelope: &Value) -> String {
    // Always inline the full envelope as JSON — LLM agents only read the text
    // content block, not structuredContent.
    //
    // For truncated payloads this means:
    //  - section_pruning: agent sees the pruned-but-still-useful data
    //  - summary_only / minimal_fallback: agent sees shape summaries + reload hint
    // In all cases the truncation metadata (strategy, hint) is part of the
    // envelope and therefore visible in the inlined JSON.
    to_pretty_json(envelope)
}

#[cfg(test)]
fn compact_openapi_section(result: &ApiCallResult) -> Value {
    let title = result
        .body
        .get("info")
        .and_then(|info| info.get("title"))
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_string();
    let version = result
        .body
        .get("info")
        .and_then(|info| info.get("version"))
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_string();
    let endpoints = extract_openapi_endpoints(&result.body);
    let endpoint_count = endpoints.len();
    let preview = if endpoint_count > COMPACT_ENDPOINT_PREVIEW_MAX_ITEMS {
        endpoints
            .into_iter()
            .take(COMPACT_ENDPOINT_PREVIEW_MAX_ITEMS)
            .collect::<Vec<_>>()
    } else {
        endpoints
    };
    json!({
        "ok": result.is_success(),
        "status": result.status,
        "summary": {
            "title": title,
            "version": version,
            "endpoint_count": endpoint_count
        },
        "compact_endpoints": preview,
        "compact_endpoints_truncated": endpoint_count > COMPACT_ENDPOINT_PREVIEW_MAX_ITEMS
    })
}

#[cfg(test)]
fn json_to_map(value: Value) -> Map<String, Value> {
    match value {
        Value::Object(map) => map,
        _ => Map::new(),
    }
}

fn enforce_tool_payload_limit(tool: &str, envelope: Value) -> Value {
    let original_bytes = serialized_json_size_bytes(&envelope);
    if original_bytes <= TOOL_ENVELOPE_MAX_BYTES {
        return envelope;
    }
    if public_tool_forbids_payload_truncation(tool) {
        return public_payload_contract_overflow(
            tool,
            &envelope,
            &envelope,
            original_bytes,
            original_bytes,
        );
    }

    let mut trimmed = envelope.clone();
    if let Some(data) = trimmed.get_mut("data").and_then(Value::as_object_mut) {
        trim_discovery_sections(data);
    }
    let trimmed_bytes = serialized_json_size_bytes(&trimmed);
    if trimmed_bytes <= TOOL_ENVELOPE_MAX_BYTES {
        let details_hint = payload_reload_hint(tool, Some(&envelope));
        annotate_truncation(
            &mut trimmed,
            "section_pruning",
            original_bytes,
            trimmed_bytes,
            details_hint,
        );
        return trimmed;
    }

    let status = envelope
        .get("status")
        .cloned()
        .unwrap_or_else(|| Value::String("complete".to_string()));
    let mut fallback = json!({
        "status": status,
        "phase": "final",
        "tool": tool,
        "truncated": true,
        "truncation": {
            "reason": "payload_size_limit",
            "strategy": "summary_only",
            "limit_bytes": TOOL_ENVELOPE_MAX_BYTES,
            "original_bytes": original_bytes,
            "details_hint": payload_reload_hint(tool, Some(&envelope))
        }
    });
    if let Some(data) = envelope.get("data") {
        fallback["data_summary"] = summarize_json_shape(data);
    }
    if let Some(error) = envelope.get("error") {
        fallback["error_summary"] = summarize_json_shape(error);
    }

    if serialized_json_size_bytes(&fallback) > TOOL_ENVELOPE_MAX_BYTES {
        json!({
            "status": status,
            "phase": "final",
            "tool": tool,
            "truncated": true,
            "truncation": {
                "reason": "payload_size_limit",
                "strategy": "minimal_fallback",
                "limit_bytes": TOOL_ENVELOPE_MAX_BYTES,
                "original_bytes": original_bytes,
                "details_hint": payload_reload_hint(tool, Some(&envelope))
            }
        })
    } else {
        fallback
    }
}

fn build_events_list_pagination_next_call(
    event_type: Option<&str>,
    since: Option<&str>,
    until: Option<&str>,
    limit: Option<u64>,
    has_more: bool,
    next_cursor: Option<&str>,
) -> Value {
    if !has_more {
        return Value::Null;
    }
    let Some(cursor) = next_cursor else {
        return Value::Null;
    };

    let mut arguments = json!({
        "cursor": cursor
    });
    if let Some(value) = event_type {
        arguments["event_type"] = Value::String(value.to_string());
    }
    if let Some(value) = since {
        arguments["since"] = Value::String(value.to_string());
    }
    if let Some(value) = until {
        arguments["until"] = Value::String(value.to_string());
    }
    if let Some(value) = limit {
        arguments["limit"] = json!(value);
    }
    json!({
        "tool": "kura_events_list",
        "arguments": arguments
    })
}

fn annotate_truncation(
    envelope: &mut Value,
    strategy: &str,
    original_bytes: usize,
    remaining_bytes: usize,
    details_hint: String,
) {
    if let Some(obj) = envelope.as_object_mut() {
        obj.insert("truncated".to_string(), Value::Bool(true));
        obj.insert(
            "truncation".to_string(),
            json!({
                "reason": "payload_size_limit",
                "strategy": strategy,
                "limit_bytes": TOOL_ENVELOPE_MAX_BYTES,
                "original_bytes": original_bytes,
                "remaining_bytes": remaining_bytes,
                "details_hint": details_hint
            }),
        );
    }
}

fn trim_discovery_sections(data: &mut Map<String, Value>) {
    for section in ["openapi", "system_config", "agent_capabilities"] {
        let Some(section_value) = data.get_mut(section) else {
            continue;
        };
        let Some(section_obj) = section_value.as_object_mut() else {
            continue;
        };

        if let Some(body) = section_obj.remove("body") {
            section_obj.insert("body_omitted".to_string(), Value::Bool(true));
            section_obj.insert("body_summary".to_string(), summarize_json_shape(&body));
        }
        if let Some(headers) = section_obj.remove("headers") {
            section_obj.insert("headers_omitted".to_string(), Value::Bool(true));
            section_obj.insert(
                "headers_summary".to_string(),
                summarize_json_shape(&headers),
            );
        }
        if let Some(endpoints) = section_obj
            .get_mut("compact_endpoints")
            .and_then(Value::as_array_mut)
        {
            let total = endpoints.len();
            if total > COMPACT_ENDPOINT_PREVIEW_MAX_ITEMS {
                endpoints.truncate(COMPACT_ENDPOINT_PREVIEW_MAX_ITEMS);
                section_obj.insert("compact_endpoints_truncated".to_string(), Value::Bool(true));
                section_obj.insert("compact_endpoints_total".to_string(), json!(total));
            }
        }
    }
}

fn summarize_json_shape(value: &Value) -> Value {
    match value {
        Value::Object(map) => {
            let mut sample_keys = map.keys().take(20).cloned().collect::<Vec<_>>();
            sample_keys.sort();
            json!({
                "omitted": true,
                "type": "object",
                "key_count": map.len(),
                "sample_keys": sample_keys
            })
        }
        Value::Array(items) => json!({
            "omitted": true,
            "type": "array",
            "item_count": items.len()
        }),
        Value::String(s) => json!({
            "omitted": true,
            "type": "string",
            "char_count": s.chars().count()
        }),
        Value::Number(_) => json!({
            "omitted": true,
            "type": "number"
        }),
        Value::Bool(_) => json!({
            "omitted": true,
            "type": "boolean"
        }),
        Value::Null => json!({
            "omitted": true,
            "type": "null"
        }),
    }
}

fn serialized_json_size_bytes(value: &Value) -> usize {
    serde_json::to_vec(value)
        .map(|bytes| bytes.len())
        .unwrap_or(usize::MAX)
}

fn payload_reload_hint(_tool: &str, envelope: Option<&Value>) -> String {
    let authoritative_training_memory_hint = envelope
        .and_then(|value| value.pointer("/request/path"))
        .and_then(Value::as_str)
        .filter(|path| path.contains("/v1/projections/authoritative_training_memory"))
        .map(|_| {
            "Use read_query with source_id=activity_session_history and subject=day_history for compact day recall, subject=session_history for compact session rows, subject=session_detail for one exact saved session, subject=block_history with query for bounded programmable activity reads, or subject=activity_timeline/activity_summary for ordered points and totals. Continue with the returned cursor when has_more=true instead of retrying removed legacy training projections.".to_string()
        });
    if let Some(hint) = authoritative_training_memory_hint {
        return hint;
    }
    let non_training_history_hint = envelope
        .and_then(|value| value.pointer("/request/path"))
        .and_then(Value::as_str)
        .and_then(|path| {
            if path.contains("/v1/projections/performance_tests") {
                Some("Use read_query with source_id=performance_tests and read_kind=aggregate for a compact benchmark summary, read_kind=list to discover saved series and entries, read_kind=timeline for ordered benchmark points, or read_kind=lookup for one exact saved test entry. Continue with the returned cursor when has_more=true instead of retrying legacy benchmark shadows.".to_string())
            } else if path.contains("/v1/projections/recovery") {
                Some("Use read_query with source_id=recovery and read_kind=list for compact recovery history, then switch to read_kind=lookup with entry_id or date for one exact recovery day. Continue with the returned cursor when has_more=true instead of retrying legacy recovery projections.".to_string())
            } else if path.contains("/v1/projections/nutrition") {
                Some("Use kura_projection_get(projection_type=nutrition, key=index) for a compact nutrition history index, kura_projection_list(projection_type=nutrition, limit=25) to page lookup keys, then kura_projection_get(projection_type=nutrition, key=<lookup_key>) for exact nutrition detail. Inspect history_truth_coverage and keep the answer explicitly unknown when coverage_status is bounded_gap.".to_string())
            } else if path.contains("/v1/projections/supplements") {
                Some("Use kura_projection_get(projection_type=supplements, key=index) for a compact supplement history index, kura_projection_list(projection_type=supplements, limit=25) to page lookup keys, then kura_projection_get(projection_type=supplements, key=<lookup_key>) for exact supplement detail. Inspect history_truth_coverage and keep the answer explicitly unknown when coverage_status is bounded_gap.".to_string())
            } else {
                None
            }
        });
    if let Some(hint) = non_training_history_hint {
        return hint;
    }

    "Retry with narrower scope or pagination, then request follow-up chunks for full detail."
        .to_string()
}

fn mcp_install_channel_header_value() -> String {
    std::env::var("KURA_MCP_INSTALL_CHANNEL")
        .ok()
        .map(|value| value.trim().to_ascii_lowercase())
        .filter(|value| !value.is_empty())
        .unwrap_or_else(|| "cargo".to_string())
}

#[cfg(test)]
fn extract_agent_user_notices(body: Option<&Value>) -> Vec<Value> {
    body.and_then(|v| v.get("user_notices"))
        .and_then(Value::as_array)
        .map(|items| {
            items
                .iter()
                .filter_map(|item| item.as_object())
                .filter(|item| {
                    item.get("relay_policy")
                        .and_then(Value::as_str)
                        .map(|value| value == "must_append_once")
                        .unwrap_or(false)
                })
                .map(|item| Value::Object(item.clone()))
                .collect::<Vec<_>>()
        })
        .unwrap_or_default()
}

fn extract_agent_user_notice_ack_ids(body: Option<&Value>) -> Vec<String> {
    body.and_then(|v| v.get("user_notices"))
        .and_then(Value::as_array)
        .map(|items| {
            items
                .iter()
                .filter_map(|item| item.as_object())
                .filter_map(|item| item.get("notice_id").and_then(Value::as_str))
                .map(str::trim)
                .filter(|value| is_valid_notice_ack_id(value))
                .map(ToString::to_string)
                .collect::<Vec<_>>()
        })
        .unwrap_or_default()
}

fn is_valid_notice_ack_id(raw: &str) -> bool {
    let trimmed = raw.trim();
    if trimmed.is_empty() || trimmed.len() > 200 {
        return false;
    }
    trimmed
        .chars()
        .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, ':' | '_' | '-' | '.'))
}

fn queue_pending_notice_ack_ids(session_id: &str, notice_ids: Vec<String>) {
    if notice_ids.is_empty() {
        return;
    }
    let mut cache = PENDING_NOTICE_ACK_IDS_BY_SESSION
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    let cutoff = Instant::now() - Duration::from_secs(CONTEXT_SESSION_TTL_SECS);
    cache.retain(|_, state| state.last_seen > cutoff);
    let entry = cache
        .entry(session_id.to_string())
        .or_insert_with(|| PendingNoticeAckState {
            last_seen: Instant::now(),
            notice_ids: BTreeSet::new(),
        });
    entry.last_seen = Instant::now();
    for notice_id in notice_ids {
        if entry.notice_ids.len() >= KURA_NOTICE_ACK_MAX_IDS {
            break;
        }
        entry.notice_ids.insert(notice_id);
    }
}

fn pending_notice_ack_header_value(session_id: &str) -> Option<String> {
    let mut cache = PENDING_NOTICE_ACK_IDS_BY_SESSION
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    let cutoff = Instant::now() - Duration::from_secs(CONTEXT_SESSION_TTL_SECS);
    cache.retain(|_, state| state.last_seen > cutoff);
    let entry = cache.get_mut(session_id)?;
    entry.last_seen = Instant::now();
    let joined = entry
        .notice_ids
        .iter()
        .take(KURA_NOTICE_ACK_MAX_IDS)
        .cloned()
        .collect::<Vec<_>>()
        .join(",");
    if joined.is_empty() {
        None
    } else {
        Some(joined)
    }
}

#[cfg(test)]
fn build_user_notice_relay(body: Option<&Value>) -> Value {
    let notices = extract_agent_user_notices(body);
    if notices.is_empty() {
        return json!({
            "status": "none",
            "notices": []
        });
    }

    let mut append_parts = Vec::<String>::new();
    for notice in &notices {
        let severity_label = match notice
            .get("severity")
            .and_then(Value::as_str)
            .map(str::trim)
            .unwrap_or_default()
        {
            "blocking" => "Blocking Kura notice",
            "important" => "Important Kura notice",
            _ => "Kura notice",
        };
        let message_short = notice
            .get("message_short")
            .and_then(Value::as_str)
            .map(str::trim)
            .filter(|value| !value.is_empty());
        let upgrade_command = notice
            .get("upgrade_command")
            .and_then(Value::as_str)
            .map(str::trim)
            .filter(|value| !value.is_empty());
        match (message_short, upgrade_command) {
            (Some(message), Some(command)) => {
                append_parts.push(format!("{severity_label}: {message} Update: {command}"));
            }
            (Some(message), None) => append_parts.push(format!("{severity_label}: {message}")),
            (None, Some(command)) => {
                append_parts.push(format!("{severity_label}: Update: {command}"))
            }
            (None, None) => {}
        }
    }
    let append_to_user_reply = append_parts.join(" ");

    json!({
        "status": "relay_required",
        "must_append_once": true,
        "append_to_user_reply": append_to_user_reply,
        "notices": notices,
    })
}

fn build_read_query_tool_envelope(
    response: &ApiCallResult,
    payload_budget_retry_applied: bool,
) -> Value {
    let response_body = response.body.clone();
    json!({
        "request": {
            "path": READ_QUERY_V4_PATH,
            "namespace": "read_harness_vnext",
            "surface": "read_query"
        },
        "response": response.to_value(),
        "contract": {
            "harness_profile": "vnext",
            "read_contract_ref": "public_agent_read_contract_v1",
            "programmable_kernel_contract_ref": "public_agent_programmable_read_kernel_contract_v1",
            "continuation_contract_ref": "public_agent_read_continuation_contract_v1",
            "analysis_handoff_contract_ref": "public_agent_read_analysis_handoff_contract_v1",
            "payload_budget_retry_applied": payload_budget_retry_applied,
            "continuation": build_read_query_continuation_contract(&response_body)
        }
    })
}

fn build_read_query_continuation_contract(body: &Value) -> Value {
    let Some(continuation) = body.get("continuation").filter(|value| !value.is_null()) else {
        return json!({
            "status": "none",
        });
    };

    if !continuation
        .get("has_more")
        .and_then(Value::as_bool)
        .unwrap_or(false)
    {
        return json!({
            "status": "none",
        });
    }

    json!({
        "status": "continue_on_same_tool",
        "same_tool": "read_query",
        "mode": continuation.get("mode").cloned().unwrap_or(Value::Null),
        "cursor": continuation.get("cursor").cloned().unwrap_or(Value::Null),
        "result_handle": continuation.get("result_handle").cloned().unwrap_or(Value::Null),
        "returned_count": continuation.get("returned_count").cloned().unwrap_or(Value::Null),
        "remaining_count": continuation.get("remaining_count").cloned().unwrap_or(Value::Null),
        "reason_code": continuation.get("reason_code").cloned().unwrap_or(Value::Null),
        "next_request": continuation.get("next_request").cloned().unwrap_or(Value::Null),
        "agent_instruction": "Call read_query again with the returned continuation payload and keep the same bounded query program.",
    })
}

#[cfg(test)]
fn write_contract_surface(body: &Value) -> Value {
    json!({
        "verification": body.get("verification").cloned().unwrap_or(Value::Null),
        "claim_guard": body.get("claim_guard").cloned().unwrap_or(Value::Null),
        "save_state": body.get("save_state").cloned().unwrap_or(Value::Null),
        "workflow_gate": body.get("workflow_gate").cloned().unwrap_or(Value::Null),
        "persist_intent": public_persist_intent_surface(body),
        "post_task_reflection": body.get("post_task_reflection").cloned().unwrap_or(Value::Null),
        "user_notices": body.get("user_notices").cloned().unwrap_or(Value::Null)
    })
}

#[cfg(test)]
fn public_persist_intent_surface(body: &Value) -> Value {
    let Some(persist_intent) = body.get("persist_intent").and_then(Value::as_object) else {
        return Value::Null;
    };

    let mode = persist_intent
        .get("mode")
        .and_then(Value::as_str)
        .unwrap_or_default();
    let user_prompt = persist_intent
        .get("user_prompt")
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty());
    let requires_user_action = mode.eq_ignore_ascii_case("ask_first") || user_prompt.is_some();
    if !requires_user_action {
        return Value::Null;
    }

    json!({
        "mode": persist_intent.get("mode").cloned().unwrap_or(Value::Null),
        "status_label": persist_intent.get("status_label").cloned().unwrap_or(Value::Null),
        "user_prompt": persist_intent.get("user_prompt").cloned().unwrap_or(Value::Null),
        "reason_codes": persist_intent
            .get("reason_codes")
            .cloned()
            .unwrap_or_else(|| json!([]))
    })
}

fn capability_profile_from_negotiation(
    result: Result<ApiCallResult, ToolError>,
) -> CapabilityProfile {
    let result = match result {
        Ok(result) => result,
        Err(err) => {
            return CapabilityProfile::fallback(
                "capability_negotiation_request_failed",
                vec![err.message],
                None,
            );
        }
    };

    if !result.is_success() {
        return CapabilityProfile::fallback(
            format!("capability_negotiation_http_{}", result.status),
            vec![format!(
                "GET /v1/agent/capabilities returned status {}. Continuing with preferred contract defaults.",
                result.status
            )],
            Some(result.body),
        );
    }

    let read_endpoint = result
        .body
        .get("preferred_read_endpoint")
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .unwrap_or("/v1/agent/context");
    let startup_endpoint = result
        .body
        .get("preferred_startup_endpoint")
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .unwrap_or("/v1/agent/context/section-index");
    let section_fetch_endpoint = result
        .body
        .get("preferred_section_fetch_endpoint")
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .unwrap_or("/v1/agent/context/section-fetch");

    if !read_endpoint.starts_with('/')
        || !startup_endpoint.starts_with('/')
        || !section_fetch_endpoint.starts_with('/')
    {
        return CapabilityProfile::fallback(
            "capability_negotiation_invalid_endpoints",
            vec![
                "Capabilities manifest contains non-path preferred endpoints; continuing with preferred contract defaults."
                    .to_string(),
            ],
            Some(result.body),
        );
    }

    let mut warnings = Vec::new();
    if let Some(min_mcp_version) = result
        .body
        .get("min_mcp_version")
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|v| !v.is_empty())
    {
        if min_mcp_version != "not_implemented" {
            match is_version_older(env!("CARGO_PKG_VERSION"), min_mcp_version) {
                Some(true) => {
                    return CapabilityProfile::fallback(
                        "capability_negotiation_version_mismatch",
                        vec![format!(
                            "MCP version {} is older than required min_mcp_version {}. Continuing with preferred contract defaults.",
                            env!("CARGO_PKG_VERSION"),
                            min_mcp_version
                        )],
                        Some(result.body),
                    );
                }
                Some(false) => {}
                None => warnings.push(format!(
                    "Could not parse min_mcp_version '{}' as semver. Continuing with preferred contract.",
                    min_mcp_version
                )),
            }
        }
    }

    CapabilityProfile::preferred(
        read_endpoint.to_string(),
        startup_endpoint.to_string(),
        section_fetch_endpoint.to_string(),
        result.body,
        warnings,
    )
}

fn is_version_older(current: &str, minimum: &str) -> Option<bool> {
    let current = parse_semver_triplet(current)?;
    let minimum = parse_semver_triplet(minimum)?;
    Some(current < minimum)
}

fn parse_semver_triplet(raw: &str) -> Option<(u64, u64, u64)> {
    let clean = raw.trim().trim_start_matches('v');
    let base = clean.split_once('-').map(|(base, _)| base).unwrap_or(clean);
    let mut parts = base.split('.');
    let major = parts.next()?.parse::<u64>().ok()?;
    let minor = parts.next()?.parse::<u64>().ok()?;
    let patch = parts.next()?.parse::<u64>().ok()?;
    Some((major, minor, patch))
}

fn parse_http_method(raw: &str) -> Result<Method, ToolError> {
    match raw.trim().to_uppercase().as_str() {
        "GET" => Ok(Method::GET),
        "POST" => Ok(Method::POST),
        "PUT" => Ok(Method::PUT),
        "PATCH" => Ok(Method::PATCH),
        "DELETE" => Ok(Method::DELETE),
        "HEAD" => Ok(Method::HEAD),
        "OPTIONS" => Ok(Method::OPTIONS),
        _ => Err(ToolError::new(
            "validation_failed",
            format!("Unsupported HTTP method '{raw}'"),
        )
        .with_field("method")),
    }
}

fn normalize_api_path(raw: &str) -> Result<String, ToolError> {
    let trimmed = raw.trim();
    if trimmed.is_empty() {
        return Err(
            ToolError::new("validation_failed", "API path must not be empty").with_field("path"),
        );
    }
    if trimmed.starts_with("http://") || trimmed.starts_with("https://") {
        return Err(ToolError::new(
            "validation_failed",
            "Pass API path only (e.g. /v1/events), not full URL",
        )
        .with_field("path"));
    }
    if trimmed.starts_with('/') {
        Ok(trimmed.to_string())
    } else {
        Ok(format!("/{trimmed}"))
    }
}

fn path_requires_auth(path: &str) -> bool {
    let p = path.trim().to_lowercase();
    !(p == "/health" || p.starts_with("/api-doc/") || p.starts_with("/swagger-ui"))
}

fn is_mutating_legacy_write_bridge_path(path: &str, method: &Method) -> bool {
    if !matches!(
        *method,
        Method::POST | Method::PUT | Method::PATCH | Method::DELETE
    ) {
        return false;
    }

    matches!(
        path.trim().to_ascii_lowercase().as_str(),
        "/v1/agent/exercise-resolve" | "/v4/agent/write-event"
    ) || path.trim().to_ascii_lowercase().starts_with("/v3/agent/")
}

fn is_admin_api_path(path: &str) -> bool {
    let p = path.trim().to_lowercase();
    p == "/v1/admin" || p.starts_with("/v1/admin/")
}

fn arg_bool(args: &Map<String, Value>, key: &str, default: bool) -> Result<bool, ToolError> {
    match args.get(key) {
        None => Ok(default),
        Some(Value::Bool(v)) => Ok(*v),
        Some(_) => Err(
            ToolError::new("validation_failed", format!("'{key}' must be a boolean"))
                .with_field(key),
        ),
    }
}

fn arg_string(args: &Map<String, Value>, key: &str, default: &str) -> Result<String, ToolError> {
    match args.get(key) {
        None => Ok(default.to_string()),
        Some(Value::String(v)) => Ok(v.clone()),
        Some(_) => Err(
            ToolError::new("validation_failed", format!("'{key}' must be a string"))
                .with_field(key),
        ),
    }
}

fn required_string(args: &Map<String, Value>, key: &str) -> Result<String, ToolError> {
    let value = args.get(key).ok_or_else(|| {
        ToolError::new(
            "validation_failed",
            format!("Missing required field '{key}'"),
        )
        .with_field(key)
    })?;
    match value {
        Value::String(v) if !v.trim().is_empty() => Ok(v.clone()),
        Value::String(_) => Err(ToolError::new(
            "validation_failed",
            format!("'{key}' must not be empty"),
        )
        .with_field(key)),
        _ => Err(
            ToolError::new("validation_failed", format!("'{key}' must be a string"))
                .with_field(key),
        ),
    }
}

fn arg_optional_string(args: &Map<String, Value>, key: &str) -> Result<Option<String>, ToolError> {
    match args.get(key) {
        None | Some(Value::Null) => Ok(None),
        Some(Value::String(v)) if v.trim().is_empty() => Ok(None),
        Some(Value::String(v)) => Ok(Some(v.clone())),
        Some(_) => Err(
            ToolError::new("validation_failed", format!("'{key}' must be a string"))
                .with_field(key),
        ),
    }
}

fn arg_optional_bool(args: &Map<String, Value>, key: &str) -> Result<Option<bool>, ToolError> {
    match args.get(key) {
        None | Some(Value::Null) => Ok(None),
        Some(Value::Bool(v)) => Ok(Some(*v)),
        Some(_) => Err(
            ToolError::new("validation_failed", format!("'{key}' must be a boolean"))
                .with_field(key),
        ),
    }
}

fn arg_optional_u64(args: &Map<String, Value>, key: &str) -> Result<Option<u64>, ToolError> {
    match args.get(key) {
        None | Some(Value::Null) => Ok(None),
        Some(Value::Number(n)) => n
            .as_u64()
            .ok_or_else(|| {
                ToolError::new(
                    "validation_failed",
                    format!("'{key}' must be an unsigned integer"),
                )
                .with_field(key)
            })
            .map(Some),
        Some(_) => Err(ToolError::new(
            "validation_failed",
            format!("'{key}' must be an unsigned integer"),
        )
        .with_field(key)),
    }
}

fn arg_optional_string_array(
    args: &Map<String, Value>,
    key: &str,
) -> Result<Option<Vec<String>>, ToolError> {
    let Some(value) = args.get(key) else {
        return Ok(None);
    };
    if value.is_null() {
        return Ok(None);
    }
    let items = value.as_array().ok_or_else(|| {
        ToolError::new(
            "validation_failed",
            format!("'{key}' must be an array of strings"),
        )
        .with_field(key)
    })?;
    let mut out = Vec::with_capacity(items.len());
    for item in items {
        let text = item.as_str().ok_or_else(|| {
            ToolError::new(
                "validation_failed",
                format!("'{key}' items must be strings"),
            )
            .with_field(key)
        })?;
        let normalized = text.trim();
        if !normalized.is_empty() {
            out.push(normalized.to_string());
        }
    }
    Ok(Some(out))
}

fn arg_optional_object(
    args: &Map<String, Value>,
    key: &str,
) -> Result<Option<Map<String, Value>>, ToolError> {
    match args.get(key) {
        None | Some(Value::Null) => Ok(None),
        Some(Value::Object(map)) => Ok(Some(map.clone())),
        Some(_) => Err(
            ToolError::new("validation_failed", format!("'{key}' must be an object"))
                .with_field(key),
        ),
    }
}

fn parse_uuid_string(value: &str, field: &str) -> Result<Uuid, ToolError> {
    Uuid::parse_str(value).map_err(|_| {
        ToolError::new(
            "validation_failed",
            format!("'{field}' must be a valid UUID"),
        )
        .with_field(field)
    })
}

fn parse_query_pairs_from_args(
    args: &Map<String, Value>,
) -> Result<Vec<(String, String)>, ToolError> {
    let query = args.get("query").filter(|value| !value.is_null());
    let query_entries = args.get("query_entries").filter(|value| !value.is_null());
    if query.is_some() && query_entries.is_some() {
        return Err(ToolError::new(
            "validation_failed",
            "Use either 'query' or 'query_entries', not both.",
        ));
    }
    if let Some(query_entries) = query_entries {
        return parse_query_pairs(Some(query_entries), "query_entries");
    }
    parse_query_pairs(query, "query")
}

fn parse_query_pairs(
    query_value: Option<&Value>,
    field: &str,
) -> Result<Vec<(String, String)>, ToolError> {
    let Some(query_value) = query_value else {
        return Ok(Vec::new());
    };

    match query_value {
        Value::Object(map) => {
            let mut out = Vec::with_capacity(map.len());
            for (k, v) in map {
                if v.is_null() {
                    continue;
                }
                out.push((k.clone(), scalar_to_string(v, field)?));
            }
            Ok(out)
        }
        Value::Array(items) => {
            let mut out = Vec::with_capacity(items.len());
            for (index, item) in items.iter().enumerate() {
                let obj = item.as_object().ok_or_else(|| {
                    ToolError::new(
                        "validation_failed",
                        format!("{field}[{index}] must be an object with key/value"),
                    )
                    .with_field(field)
                })?;
                let key = obj
                    .get("key")
                    .and_then(Value::as_str)
                    .filter(|s| !s.trim().is_empty())
                    .ok_or_else(|| {
                        ToolError::new(
                            "validation_failed",
                            format!("{field}[{index}].key must be a non-empty string"),
                        )
                        .with_field(field)
                    })?;
                let value = obj.get("value").ok_or_else(|| {
                    ToolError::new(
                        "validation_failed",
                        format!("{field}[{index}].value is required"),
                    )
                    .with_field(field)
                })?;
                out.push((key.to_string(), scalar_to_string(value, field)?));
            }
            Ok(out)
        }
        _ => Err(ToolError::new(
            "validation_failed",
            format!("'{field}' must be an object map or [{{key,value}}] array"),
        )
        .with_field(field)),
    }
}

fn scalar_to_string(value: &Value, field: &str) -> Result<String, ToolError> {
    match value {
        Value::String(v) => Ok(v.clone()),
        Value::Number(v) => Ok(v.to_string()),
        Value::Bool(v) => Ok(v.to_string()),
        _ => Err(ToolError::new(
            "validation_failed",
            format!("'{field}' values must be scalar (string/number/bool)"),
        )
        .with_field(field)),
    }
}

#[cfg(test)]
const PLAN_UPDATE_VOLUME_DELTA_HIGH_IMPACT_ABS_GTE: f64 = 15.0;
#[cfg(test)]
const PLAN_UPDATE_INTENSITY_DELTA_HIGH_IMPACT_ABS_GTE: f64 = 10.0;
#[cfg(test)]
const PLAN_UPDATE_FREQUENCY_DELTA_HIGH_IMPACT_ABS_GTE: f64 = 2.0;
#[cfg(test)]
const PLAN_UPDATE_DURATION_DELTA_WEEKS_HIGH_IMPACT_ABS_GTE: f64 = 2.0;
#[cfg(test)]
const SCHEDULE_EXCEPTION_LOW_IMPACT_MAX_RANGE_DAYS: i64 = 14;

fn normalized_event_type(event: &Value) -> Option<String> {
    event
        .as_object()
        .and_then(|obj| obj.get("event_type"))
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(|value| value.to_lowercase())
}

fn contains_plan_or_schedule_writes(events: &[Value]) -> bool {
    events.iter().any(|event| {
        normalized_event_type(event)
            .map(|event_type| {
                event_type.starts_with("training_plan.")
                    || event_type.starts_with("training_schedule.exception.")
            })
            .unwrap_or(false)
    })
}

// Back-compat helper name kept for architecture contract grep and downstream references.
#[allow(dead_code)]
fn contains_plan_writes(events: &[Value]) -> bool {
    contains_plan_or_schedule_writes(events)
}

#[cfg(test)]
fn is_canonical_logging_event_type(event_type: &str) -> bool {
    matches!(
        event_type.trim().to_lowercase().as_str(),
        "set.logged" | "performance_test.logged" | "session.logged" | "session.completed"
    )
}

#[cfg(test)]
fn is_core_repair_event_type(event_type: &str) -> bool {
    matches!(
        event_type.trim().to_lowercase().as_str(),
        "set.corrected" | "event.retracted"
    )
}

#[cfg(test)]
fn contains_canonical_logging_writes(events: &[Value]) -> bool {
    events.iter().any(|event| {
        normalized_event_type(event)
            .is_some_and(|event_type| is_canonical_logging_event_type(&event_type))
    })
}

#[cfg(test)]
fn contains_core_repair_writes(events: &[Value]) -> bool {
    events.iter().any(|event| {
        normalized_event_type(event)
            .is_some_and(|event_type| is_core_repair_event_type(&event_type))
    })
}

#[cfg(test)]
fn is_always_high_impact_event_type(event_type: &str) -> bool {
    matches!(
        event_type.trim().to_lowercase().as_str(),
        "training_plan.created"
            | "training_plan.archived"
            | "projection_rule.created"
            | "projection_rule.archived"
            | "weight_target.set"
            | "sleep_target.set"
            | "nutrition_target.set"
            | "workflow.profile_completion.closed"
            | "workflow.profile_completion.override_granted"
            | "workflow.profile_completion.aborted"
            | "workflow.profile_completion.restarted"
    )
}

#[cfg(test)]
fn read_abs_f64(value: Option<&Value>) -> Option<f64> {
    let raw = value?;
    if let Some(number) = raw.as_f64() {
        return Some(number.abs());
    }
    if let Some(number) = raw.as_i64() {
        return Some((number as f64).abs());
    }
    if let Some(number) = raw.as_u64() {
        return Some((number as f64).abs());
    }
    raw.as_str()
        .and_then(|text| text.trim().parse::<f64>().ok())
        .map(f64::abs)
}

#[cfg(test)]
fn read_plan_delta_abs(data: &Value, keys: &[&str]) -> Option<f64> {
    for key in keys {
        if let Some(number) = read_abs_f64(data.get(*key)) {
            return Some(number);
        }
        if let Some(number) = read_abs_f64(data.get("delta").and_then(|delta| delta.get(*key))) {
            return Some(number);
        }
    }
    None
}

#[cfg(test)]
fn read_bool_like(value: Option<&Value>) -> Option<bool> {
    let raw = value?;
    if let Some(boolean) = raw.as_bool() {
        return Some(boolean);
    }
    if let Some(number) = raw.as_i64() {
        return match number {
            0 => Some(false),
            1 => Some(true),
            _ => None,
        };
    }
    raw.as_str()
        .and_then(|text| match text.trim().to_lowercase().as_str() {
            "true" | "yes" | "ja" | "1" | "on" | "active" => Some(true),
            "false" | "no" | "nein" | "0" | "off" | "inactive" => Some(false),
            _ => None,
        })
}

#[cfg(test)]
fn parse_local_date_value(value: Option<&Value>) -> Option<chrono::NaiveDate> {
    value
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .and_then(|value| chrono::NaiveDate::parse_from_str(value, "%Y-%m-%d").ok())
}

#[cfg(test)]
fn selector_has_explicit_occurrence_anchor(selector: &Value) -> bool {
    selector
        .get("occurrence_id")
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .is_some()
        || selector
            .get("occurrence_ids")
            .and_then(Value::as_array)
            .map(|values| {
                values.iter().any(|value| {
                    value
                        .as_str()
                        .map(str::trim)
                        .filter(|raw| !raw.is_empty())
                        .is_some()
                })
            })
            .unwrap_or(false)
}

#[cfg(test)]
fn selector_has_bounded_temporal_anchor(selector: &Value) -> bool {
    if selector_has_explicit_occurrence_anchor(selector) {
        return true;
    }
    if parse_local_date_value(selector.get("local_date").or_else(|| selector.get("date"))).is_some()
    {
        return true;
    }
    if selector
        .get("local_dates")
        .and_then(Value::as_array)
        .map(|values| {
            values
                .iter()
                .any(|value| parse_local_date_value(Some(value)).is_some())
        })
        .unwrap_or(false)
    {
        return true;
    }
    if parse_local_date_value(selector.get("week_of")).is_some() {
        return true;
    }

    let date_range = selector
        .get("date_range")
        .or_else(|| selector.get("between"))
        .unwrap_or(&Value::Null);
    let start = parse_local_date_value(date_range.get("start").or_else(|| date_range.get("from")));
    let end = parse_local_date_value(date_range.get("end").or_else(|| date_range.get("to")));
    match (start, end) {
        (Some(start), Some(end)) if end >= start => {
            (end - start).num_days() <= SCHEDULE_EXCEPTION_LOW_IMPACT_MAX_RANGE_DAYS
        }
        _ => false,
    }
}

#[cfg(test)]
fn schedule_exception_scope_is_high_impact(data: &Value) -> bool {
    let scope_value = data
        .get("change_scope")
        .or_else(|| data.get("update_scope"))
        .or_else(|| {
            data.get("scope")
                .and_then(|scope| scope.get("change_scope"))
        })
        .or_else(|| data.get("scope").and_then(|scope| scope.get("scope")))
        .and_then(Value::as_str)
        .map(|raw| raw.trim().to_lowercase());
    if matches!(
        scope_value.as_deref(),
        Some(
            "bulk"
                | "future_block"
                | "full_rewrite"
                | "template_rewrite"
                | "replace_future_schedule"
                | "mesocycle_reset"
                | "phase_shift"
        )
    ) {
        return true;
    }

    for key in ["days_affected", "occurrences_affected"] {
        if read_abs_f64(data.get("scope").and_then(|scope| scope.get(key))).unwrap_or(0.0)
            > SCHEDULE_EXCEPTION_LOW_IMPACT_MAX_RANGE_DAYS as f64
        {
            return true;
        }
    }
    if read_abs_f64(
        data.get("scope")
            .and_then(|scope| scope.get("weeks_affected")),
    )
    .unwrap_or(0.0)
        > 2.0
    {
        return true;
    }
    false
}

#[cfg(test)]
fn training_schedule_exception_is_high_impact(event_type: &str, data: &Value) -> bool {
    if read_bool_like(data.get("requires_explicit_confirmation")).unwrap_or(false)
        || read_bool_like(data.get("rewrite_template")).unwrap_or(false)
        || read_bool_like(data.get("replace_future_schedule")).unwrap_or(false)
        || read_bool_like(data.get("replace_entire_weekly_template")).unwrap_or(false)
        || read_bool_like(data.get("clear_all")).unwrap_or(false)
        || schedule_exception_scope_is_high_impact(data)
    {
        return true;
    }

    match event_type {
        "training_schedule.exception.cleared" => false,
        "training_schedule.exception.upsert" => {
            let selector = data.get("selector").unwrap_or(&Value::Null);
            !selector_has_bounded_temporal_anchor(selector)
        }
        _ => true,
    }
}

#[cfg(test)]
fn training_plan_update_is_high_impact(data: &Value) -> bool {
    let scope = data
        .get("change_scope")
        .or_else(|| data.get("update_scope"))
        .and_then(Value::as_str)
        .map(|raw| raw.trim().to_lowercase());
    if matches!(
        scope.as_deref(),
        Some(
            "full_rewrite" | "structural" | "major_adjustment" | "mesocycle_reset" | "phase_shift"
        )
    ) {
        return true;
    }

    if data
        .get("replace_entire_plan")
        .and_then(Value::as_bool)
        .unwrap_or(false)
        || data
            .get("archive_previous_plan")
            .and_then(Value::as_bool)
            .unwrap_or(false)
        || data
            .get("requires_explicit_confirmation")
            .and_then(Value::as_bool)
            .unwrap_or(false)
    {
        return true;
    }

    let volume_delta = read_plan_delta_abs(
        data,
        &[
            "volume_delta_pct",
            "planned_volume_delta_pct",
            "total_volume_delta_pct",
        ],
    )
    .unwrap_or(0.0);
    if volume_delta >= PLAN_UPDATE_VOLUME_DELTA_HIGH_IMPACT_ABS_GTE {
        return true;
    }

    let intensity_delta = read_plan_delta_abs(
        data,
        &[
            "intensity_delta_pct",
            "rir_delta",
            "rpe_delta",
            "effort_delta_pct",
        ],
    )
    .unwrap_or(0.0);
    if intensity_delta >= PLAN_UPDATE_INTENSITY_DELTA_HIGH_IMPACT_ABS_GTE {
        return true;
    }

    let frequency_delta = read_plan_delta_abs(
        data,
        &["frequency_delta_per_week", "sessions_per_week_delta"],
    )
    .unwrap_or(0.0);
    if frequency_delta >= PLAN_UPDATE_FREQUENCY_DELTA_HIGH_IMPACT_ABS_GTE {
        return true;
    }

    let duration_delta = read_plan_delta_abs(
        data,
        &["cycle_length_weeks_delta", "plan_duration_weeks_delta"],
    )
    .unwrap_or(0.0);
    duration_delta >= PLAN_UPDATE_DURATION_DELTA_WEEKS_HIGH_IMPACT_ABS_GTE
}

#[cfg(test)]
fn is_high_impact_event(event: &Value) -> bool {
    let Some(event_type) = normalized_event_type(event) else {
        return false;
    };
    if event_type == "training_plan.updated" {
        return event
            .as_object()
            .and_then(|obj| obj.get("data"))
            .is_some_and(training_plan_update_is_high_impact);
    }
    if event_type == "training_schedule.exception.upsert"
        || event_type == "training_schedule.exception.cleared"
    {
        return event
            .as_object()
            .and_then(|obj| obj.get("data"))
            .is_some_and(|data| training_schedule_exception_is_high_impact(&event_type, data));
    }
    is_always_high_impact_event_type(&event_type)
}

#[cfg(test)]
fn has_high_impact_events(events: &[Value]) -> bool {
    events.iter().any(is_high_impact_event)
}

#[cfg(test)]
fn confirmation_token_from_value(value: &Value) -> Option<String> {
    value
        .as_object()
        .and_then(|obj| obj.get("confirmation_token"))
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|token| !token.is_empty())
        .map(str::to_string)
}

#[cfg(test)]
fn response_confirmation_token(body: &Value) -> Option<String> {
    body.get("received").and_then(confirmation_token_from_value)
}

#[cfg(test)]
fn prune_high_impact_confirmation_replay_cache(
    cache: &mut HashMap<String, HighImpactConfirmationReplayEntry>,
) {
    let now = Instant::now();
    cache.retain(|_, entry| {
        now.duration_since(entry.last_seen)
            <= Duration::from_secs(HIGH_IMPACT_CONFIRMATION_REPLAY_CACHE_TTL_SECS)
    });
}

#[cfg(test)]
fn remember_high_impact_confirmation_replay(path: &str, body: &Value, response_body: &Value) {
    let Some(token) = response_confirmation_token(response_body) else {
        return;
    };
    let mut cache = HIGH_IMPACT_CONFIRMATION_REPLAY_CACHE
        .lock()
        .expect("high-impact replay cache lock poisoned");
    prune_high_impact_confirmation_replay_cache(&mut cache);
    cache.insert(
        token,
        HighImpactConfirmationReplayEntry {
            last_seen: Instant::now(),
            path: path.to_string(),
            body: body.clone(),
        },
    );
}

#[cfg(test)]
fn replay_body_for_high_impact_confirmation(
    token: &str,
    confirmation: &Value,
) -> Option<(String, Value)> {
    let mut cache = HIGH_IMPACT_CONFIRMATION_REPLAY_CACHE
        .lock()
        .expect("high-impact replay cache lock poisoned");
    prune_high_impact_confirmation_replay_cache(&mut cache);
    let entry = cache.get_mut(token)?;
    entry.last_seen = Instant::now();
    let mut body = entry.body.clone();
    body["high_impact_confirmation"] = confirmation.clone();
    Some((entry.path.clone(), body))
}

#[cfg(test)]
fn plan_write_requires_family_write_error(
    requested_mode: WriteMode,
    capability_mode: &str,
    context: Option<Value>,
) -> ToolError {
    // Back-compat contract phrase retained for architecture spec grep:
    // "training_plan.* writes must use mode=family_write"
    let mut details = json!({
        "contract_error": "inv_plan_write_requires_family_write",
        "required_mode": "family_write",
        "requested_mode": requested_mode.as_str(),
        "capability_mode": capability_mode,
        "plan_event_prefix": "training_plan.",
        "schedule_exception_event_prefix": "training_schedule.exception."
    });
    if let Some(context) = context {
        details["context"] = context;
    }

    ToolError::new(
        "plan_write_requires_family_write",
        "training_plan.* and training_schedule.exception.* writes must use mode=family_write; legacy commit paths are blocked",
    )
    .with_field("mode")
    .with_docs_hint(
        "Use mode=family_write with read_after_write_targets. Do not use commit or legacy fallback for plan/schedule writes.",
    )
    .with_details(details)
}

#[cfg(test)]
#[derive(Debug, Clone)]
struct WriteAutoRetryPlan {
    reason_code: String,
    event_index: usize,
    patched_events: Vec<Value>,
    applied_fix: Value,
}

#[cfg(test)]
fn parse_event_index_from_error_field(field: Option<&str>, event_count: usize) -> Option<usize> {
    let field = field?;
    let marker = "events[";
    let start = field.find(marker)?;
    let rest = &field[start + marker.len()..];
    let end = rest.find(']')?;
    let index = rest[..end].parse::<usize>().ok()?;
    (index < event_count).then_some(index)
}

#[cfg(test)]
fn normalize_set_type_token(value: &str) -> Option<String> {
    let normalized = value.trim().to_ascii_lowercase();
    if normalized.is_empty() {
        return None;
    }
    for (needle, canonical) in [
        ("warmup", "warmup"),
        ("warm-up", "warmup"),
        ("backoff", "backoff"),
        ("back-off", "backoff"),
        ("amrap", "amrap"),
        ("working", "working"),
    ] {
        if normalized.contains(needle) {
            return Some(canonical.to_string());
        }
    }
    None
}

#[cfg(test)]
fn remediation_v1_payload(body: &Value) -> Option<&Value> {
    body.get("received")
        .and_then(|value| value.get("remediation_v1"))
        .filter(|value| value.is_object())
}

#[cfg(test)]
fn remediation_v1_has_example_patch(body: &Value) -> bool {
    remediation_v1_payload(body)
        .and_then(|value| value.get("example_patch"))
        .is_some_and(|value| !value.is_null())
}

#[cfg(test)]
fn remediation_v1_set_type(body: &Value) -> Option<String> {
    let from_contract = remediation_v1_payload(body)
        .and_then(|value| value.get("received"))
        .and_then(|value| value.get("mentioned_set_type"))
        .and_then(Value::as_str);
    let from_legacy = body
        .get("received")
        .and_then(|value| value.get("mentioned_set_type"))
        .and_then(Value::as_str);
    from_contract
        .or(from_legacy)
        .and_then(normalize_set_type_token)
}

#[cfg(test)]
fn derive_write_auto_retry_plan(
    response: &ApiCallResult,
    events: &[Value],
) -> Option<WriteAutoRetryPlan> {
    let api_error = response
        .body
        .get("error")
        .and_then(Value::as_str)
        .map(str::trim)
        .unwrap_or_default();
    if api_error != SET_TYPE_EXPLICIT_MENTION_API_ERROR {
        return None;
    }
    let event_index = parse_event_index_from_error_field(
        response.body.get("field").and_then(Value::as_str),
        events.len(),
    )
    .or_else(|| (events.len() == 1).then_some(0))?;
    let set_type = remediation_v1_set_type(&response.body)?;

    let mut patched_events = events.to_vec();
    let event = patched_events.get_mut(event_index)?;
    let event_type = normalized_event_type(event)?;
    if event_type != "set.logged" {
        return None;
    }
    let event_obj = event.as_object_mut()?;
    let data = event_obj.get_mut("data")?.as_object_mut()?;
    if data
        .get("set_type")
        .and_then(Value::as_str)
        .map(str::trim)
        .is_some_and(|value| !value.is_empty())
    {
        return None;
    }
    data.insert("set_type".to_string(), Value::String(set_type.clone()));

    Some(WriteAutoRetryPlan {
        reason_code: SET_TYPE_EXPLICIT_MENTION_API_ERROR.to_string(),
        event_index,
        applied_fix: json!({
            "field_path": "data.set_type",
            "set_type": set_type,
        }),
        patched_events,
    })
}

#[cfg(test)]
fn build_retry_request_for_plan(
    mode: WriteMode,
    effective_path: &str,
    request_body: &Value,
    plan: &WriteAutoRetryPlan,
) -> Option<(String, Value)> {
    match mode {
        WriteMode::Commit => {
            let mut retry_body = request_body.clone();
            let body_obj = retry_body.as_object_mut()?;
            if body_obj.contains_key("events") {
                body_obj.insert(
                    "events".to_string(),
                    Value::Array(plan.patched_events.clone()),
                );
            } else if body_obj.contains_key("event") {
                let event = plan.patched_events.first()?.clone();
                body_obj.insert("event".to_string(), event);
            } else {
                return None;
            }
            Some((effective_path.to_string(), retry_body))
        }
        WriteMode::FamilyWrite => {
            let mut retry_body = request_body.clone();
            let body_obj = retry_body.as_object_mut()?;
            body_obj.insert(
                "events".to_string(),
                Value::Array(plan.patched_events.clone()),
            );
            Some((effective_path.to_string(), retry_body))
        }
    }
}

#[cfg(test)]
#[derive(Debug, Clone)]
struct WriteApiErrorClassification {
    code: String,
    message: String,
    docs_hint: Option<String>,
    remediation_code: Option<String>,
    remediation_event_type: Option<String>,
    remediation_field_path: Option<String>,
    next_action: Option<String>,
    next_action_url: Option<String>,
}

#[cfg(test)]
fn classify_write_api_error(response: &ApiCallResult) -> WriteApiErrorClassification {
    let remediation_v1 = remediation_v1_payload(&response.body);
    let remediation_code = remediation_v1
        .and_then(|value| value.get("code"))
        .and_then(Value::as_str)
        .map(ToString::to_string);
    let remediation_event_type = remediation_v1
        .and_then(|value| value.get("event_type"))
        .and_then(Value::as_str)
        .map(ToString::to_string);
    let remediation_field_path = remediation_v1
        .and_then(|value| value.get("field_path"))
        .and_then(Value::as_str)
        .map(ToString::to_string);
    let api_error = response
        .body
        .get("error")
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(ToString::to_string);
    let docs_hint = response
        .body
        .get("docs_hint")
        .and_then(Value::as_str)
        .map(ToString::to_string);
    let next_action = response
        .body
        .get("next_action")
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(ToString::to_string);
    let next_action_url = response
        .body
        .get("next_action_url")
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(ToString::to_string);
    let message = response
        .body
        .get("message")
        .and_then(Value::as_str)
        .map(str::trim)
        .unwrap_or_default();
    let message_lc = message.to_lowercase();

    let (code, mapped_message, mapped_docs_hint, mapped_next_action) = if api_error.as_deref()
        == Some("inv_plan_write_requires_family_write")
    {
        (
            "plan_write_requires_family_write".to_string(),
            "Plan writes must use /v4/agent/update-plan; legacy event write endpoints are blocked."
                .to_string(),
            Some("Stay on update_plan and provide read_after_write_targets when you need verification.".to_string()),
            next_action.clone(),
        )
    } else if api_error.as_deref() == Some(CANONICAL_LOGGING_DIRECT_WRITE_API_ERROR) {
        (
            CANONICAL_LOGGING_DIRECT_WRITE_API_ERROR.to_string(),
            "Canonical logging writes must use task-first public write surfaces; raw event create is blocked.".to_string(),
            Some(docs_hint.clone().unwrap_or_else(|| {
                "Retry via /v4/agent/record-activity or /v4/agent/record-benchmark so Kura can enforce canonical payload and session contracts.".to_string()
            })),
            Some(
                next_action
                    .clone()
                    .unwrap_or_else(|| CANONICAL_LOGGING_DIRECT_WRITE_NEXT_ACTION.to_string()),
            ),
        )
    } else if api_error.as_deref() == Some(CANONICAL_LOGGING_UNKNOWN_FIELD_API_ERROR) {
        let canonical_retry_hint = if remediation_v1_has_example_patch(&response.body) {
            "Reuse error.details.response.received.remediation_v1.example_patch as the retry template; do not invent new fields or fall back to raw commit."
                .to_string()
        } else {
            "Use the canonical fields from event_conventions for set.logged/performance_test.logged before retrying."
                .to_string()
        };
        (
            CANONICAL_LOGGING_UNKNOWN_FIELD_API_ERROR.to_string(),
            "Canonical logging write blocked: unknown fields must be rewritten to canonical schema before commit.".to_string(),
            Some(docs_hint.clone().unwrap_or(canonical_retry_hint)),
            Some(
                next_action
                    .clone()
                    .unwrap_or_else(|| CANONICAL_LOGGING_UNKNOWN_FIELD_NEXT_ACTION.to_string()),
            ),
        )
    } else if api_error.as_deref() == Some(UNKNOWN_FIELD_HIGH_IMPACT_API_ERROR) {
        (
            "unknown_field_high_impact_blocked".to_string(),
            "High-impact write blocked: unknown fields require explicit opt-in or canonical fields before commit."
                .to_string(),
            Some(docs_hint.clone().unwrap_or_else(|| {
                "Use canonical fields from event_conventions or include metadata.context.unknown_field_opt_in.allow_new_fields. If uncertain, route details to observation.logged first.".to_string()
            })),
            Some(
                next_action
                    .clone()
                    .unwrap_or_else(|| UNKNOWN_FIELD_HIGH_IMPACT_NEXT_ACTION.to_string()),
            ),
        )
    } else if api_error.as_deref() == Some(SET_TYPE_EXPLICIT_MENTION_API_ERROR) {
        (
            SET_TYPE_EXPLICIT_MENTION_API_ERROR.to_string(),
            "set.logged write blocked: explicit warmup/backoff/amrap/working mention requires data.set_type."
                .to_string(),
            Some(docs_hint.clone().unwrap_or_else(|| {
                "Persist data.set_type whenever the same event text explicitly encodes set context."
                    .to_string()
            })),
            Some(
                next_action
                    .clone()
                    .unwrap_or_else(|| SET_TYPE_EXPLICIT_MENTION_NEXT_ACTION.to_string()),
            ),
        )
    } else if api_error.as_deref() == Some("validation_failed")
        && message_lc.contains("family_write blocked by preflight checks")
    {
        let first_blocker = response
            .body
            .pointer("/received/blockers/0/code")
            .and_then(Value::as_str)
            .map(str::to_string);
        (
            "write_preflight_blocked".to_string(),
            match first_blocker {
                Some(code) => format!("Write blocked by preflight checks (first blocker: {code})."),
                None => "Write blocked by preflight checks.".to_string(),
            },
            docs_hint.clone(),
            next_action.clone(),
        )
    } else if message_lc.contains("approval")
        && (message_lc.contains("timed out") || message_lc.contains("timeout"))
    {
        (
            "approval_timeout".to_string(),
            "Execution approval timed out before the write could run.".to_string(),
            docs_hint.or(Some(
                "Retry and approve promptly, or route execution to a non-interactive node."
                    .to_string(),
            )),
            next_action.clone(),
        )
    } else {
        (
            api_error.clone().unwrap_or_else(|| "api_error".to_string()),
            if message.is_empty() {
                format!("write_event failed with HTTP {}", response.status)
            } else {
                message.to_string()
            },
            docs_hint.clone(),
            next_action.clone(),
        )
    };

    let mapped_next_action = if mapped_next_action.is_some() {
        mapped_next_action
    } else if api_error.as_deref() == Some(UNKNOWN_FIELD_HIGH_IMPACT_API_ERROR) {
        Some(UNKNOWN_FIELD_HIGH_IMPACT_NEXT_ACTION.to_string())
    } else {
        None
    };

    let mapped_next_action = mapped_next_action.or_else(|| {
        if api_error.as_deref() == Some(UNKNOWN_FIELD_HIGH_IMPACT_API_ERROR) {
            Some(UNKNOWN_FIELD_HIGH_IMPACT_FALLBACK_ACTION.to_string())
        } else {
            None
        }
    });

    WriteApiErrorClassification {
        code,
        message: mapped_message,
        docs_hint: mapped_docs_hint,
        remediation_code,
        remediation_event_type,
        remediation_field_path,
        next_action: mapped_next_action,
        next_action_url,
    }
}

#[cfg(test)]
fn build_default_intent_handshake(
    events: &[Value],
    intent_goal: Option<&str>,
    temporal_basis: Value,
) -> Value {
    let event_types: Vec<String> = events
        .iter()
        .filter_map(|event| {
            event
                .as_object()
                .and_then(|obj| obj.get("event_type"))
                .and_then(Value::as_str)
                .map(|value| value.trim().to_lowercase())
        })
        .collect();
    let planned_action = if event_types.is_empty() {
        "apply high-impact retained-family update".to_string()
    } else {
        format!("write events: {}", event_types.join(", "))
    };
    let goal = intent_goal.unwrap_or("execute requested high-impact write safely");
    let created_at = auto_generated_intent_handshake_created_at(&temporal_basis)
        .unwrap_or_else(|| chrono::Utc::now().to_rfc3339());
    let handshake_id = auto_generated_intent_handshake_id(&event_types, goal, &temporal_basis);

    json!({
        "schema_version": "intent_handshake.v1",
        "goal": goal,
        "planned_action": planned_action,
        "assumptions": ["context and request intent are current"],
        "non_goals": ["no unrelated writes outside current task scope"],
        "impact_class": "high_impact_write",
        "success_criteria": "family_write returns verification and claim_guard for this action",
        "created_at": created_at,
        "handshake_id": handshake_id,
        "temporal_basis": temporal_basis,
    })
}

#[cfg(test)]
fn auto_generated_intent_handshake_created_at(temporal_basis: &Value) -> Option<String> {
    temporal_basis
        .as_object()
        .and_then(|obj| obj.get("context_generated_at"))
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(str::to_string)
}

#[cfg(test)]
fn auto_generated_intent_handshake_id(
    event_types: &[String],
    goal: &str,
    temporal_basis: &Value,
) -> String {
    let seed = json!({
        "event_types": event_types,
        "goal": goal,
        "temporal_basis": temporal_basis,
    })
    .to_string();
    format!("mcp-hs-{}", stable_mcp_hex_suffix(&seed))
}

#[cfg(test)]
fn stable_mcp_hex_suffix(seed: &str) -> String {
    let mut hash: u64 = 0xcbf29ce484222325;
    for byte in seed.as_bytes() {
        hash ^= u64::from(*byte);
        hash = hash.wrapping_mul(0x100000001b3);
    }
    format!("{hash:016x}")
}

fn pairs_to_json_object(pairs: &[(String, String)]) -> Value {
    let mut map = Map::new();
    for (k, v) in pairs {
        map.insert(k.clone(), Value::String(v.clone()));
    }
    Value::Object(map)
}

fn parse_response_body(bytes: &[u8]) -> Value {
    if bytes.is_empty() {
        return Value::Null;
    }
    serde_json::from_slice(bytes)
        .unwrap_or_else(|_| Value::String(String::from_utf8_lossy(bytes).to_string()))
}

#[cfg(test)]
fn extract_openapi_endpoints(spec: &Value) -> Vec<Value> {
    let mut endpoints = Vec::new();
    let Some(paths) = spec.get("paths").and_then(Value::as_object) else {
        return endpoints;
    };

    for (path, methods) in paths {
        let Some(methods) = methods.as_object() else {
            continue;
        };
        for (method, details) in methods {
            if !matches!(
                method.as_str(),
                "get" | "post" | "put" | "patch" | "delete" | "head" | "options"
            ) {
                continue;
            }
            let summary = details
                .get("summary")
                .and_then(Value::as_str)
                .unwrap_or_default();
            let requires_auth = details
                .get("security")
                .map(|security| {
                    !security
                        .as_array()
                        .map(|arr| arr.is_empty())
                        .unwrap_or(true)
                })
                .unwrap_or(false);
            endpoints.push(json!({
                "method": method.to_uppercase(),
                "path": path,
                "summary": summary,
                "auth": requires_auth
            }));
        }
    }

    endpoints.sort_by(|a, b| {
        let path_a = a.get("path").and_then(Value::as_str).unwrap_or_default();
        let path_b = b.get("path").and_then(Value::as_str).unwrap_or_default();
        let method_a = a.get("method").and_then(Value::as_str).unwrap_or_default();
        let method_b = b.get("method").and_then(Value::as_str).unwrap_or_default();
        path_a.cmp(path_b).then(method_a.cmp(method_b))
    });
    endpoints
}

fn success_response(id: Value, result: Value) -> Value {
    json!({
        "jsonrpc": "2.0",
        "id": id,
        "result": result
    })
}

fn error_response(id: Value, error: RpcError) -> Value {
    let mut payload = json!({
        "jsonrpc": "2.0",
        "id": id,
        "error": {
            "code": error.code,
            "message": error.message
        }
    });
    if let Some(data) = error.data {
        payload["error"]["data"] = data;
    }
    payload
}

async fn read_framed_json<R>(reader: &mut BufReader<R>) -> Result<Option<Value>, std::io::Error>
where
    R: tokio::io::AsyncRead + Unpin,
{
    let mut first_line = String::new();
    let bytes_read = reader.read_line(&mut first_line).await?;
    if bytes_read == 0 {
        return Ok(None);
    }

    let trimmed_first_line = first_line.trim_end_matches(['\r', '\n']);
    if !trimmed_first_line
        .to_ascii_lowercase()
        .starts_with("content-length:")
    {
        let json: Value = serde_json::from_str(trimmed_first_line).map_err(|e| {
            std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!("Invalid JSON-RPC line payload: {e}"),
            )
        })?;
        return Ok(Some(json));
    }

    let mut content_length: Option<usize> = parse_content_length_header(trimmed_first_line)?;
    loop {
        let mut line = String::new();
        let bytes_read = reader.read_line(&mut line).await?;
        if bytes_read == 0 {
            if content_length.is_none() {
                return Ok(None);
            }
            return Err(std::io::Error::new(
                std::io::ErrorKind::UnexpectedEof,
                "Unexpected EOF while reading MCP headers",
            ));
        }

        if line == "\r\n" || line == "\n" {
            break;
        }

        let line = line.trim_end_matches(['\r', '\n']);
        if line.to_ascii_lowercase().starts_with("content-length:") {
            content_length = parse_content_length_header(line)?;
        }
    }

    let content_length = content_length.ok_or_else(|| {
        std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "Missing Content-Length header",
        )
    })?;
    let mut payload = vec![0_u8; content_length];
    reader.read_exact(&mut payload).await?;

    let json: Value = serde_json::from_slice(&payload).map_err(|e| {
        std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            format!("Invalid JSON payload: {e}"),
        )
    })?;
    Ok(Some(json))
}

fn parse_content_length_header(line: &str) -> Result<Option<usize>, std::io::Error> {
    let raw_len = line
        .split_once(':')
        .map(|(_, right)| right.trim())
        .unwrap_or_default();
    let parsed = raw_len.parse::<usize>().map_err(|_| {
        std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "Invalid Content-Length header",
        )
    })?;
    Ok(Some(parsed))
}

async fn write_framed_json<W>(writer: &mut W, value: &Value) -> Result<(), std::io::Error>
where
    W: tokio::io::AsyncWrite + Unpin,
{
    let body = serde_json::to_vec(value).map_err(|e| {
        std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            format!("Failed to serialize JSON: {e}"),
        )
    })?;
    writer.write_all(&body).await?;
    writer.write_all(b"\n").await?;
    writer.flush().await?;
    Ok(())
}

fn to_pretty_json(value: &Value) -> String {
    serde_json::to_string_pretty(value).unwrap_or_else(|_| "{}".to_string())
}

#[cfg(test)]
#[path = "lib_tests.rs"]
mod tests;
