kura_cli/commands/

read.rs

use clap::{Args, Subcommand, ValueEnum};
use serde_json::{Map, Value, json};

use crate::util::{
    RawApiResponse, dry_run_enabled, emit_dry_run_request, exit_error, print_json_stderr,
    print_json_stdout, raw_api_request_json, raw_api_request_with_query,
};

const READ_SOURCE_CATALOG_SECTION: &str =
    "system_config.conventions::public_agent_read_source_catalog_v1";
const READ_SOURCE_CATALOG_ENDPOINT: &str = "/v1/system/config/section";
const READ_QUERY_ENDPOINT: &str = "/v4/agent/read-query";
const READ_QUERY_SCHEMA_VERSION: &str = "read_query_request.v1";
const READ_QUERY_HELP: &str = r#"Examples:
  kura read_query --source-id activity_session_history --read-kind list --subject day_history --filters-json '{"date_from":"2026-04-02","date_to":"2026-04-08"}' --output-shape collection
  kura read_query --source-id activity_session_history --read-kind lookup --subject session_detail --filters-json '{"session_id":"activity_session_019d77b001c472c1ae842d2dba567099"}' --limit 8 --output-shape item
  kura read_query --source-id activity_session_history --read-kind aggregate --subject block_history --filters-json '{"date_from":"2026-04-01"}' --query-json '{"group_by":"exercise_identity","aggregate":"load_kg_max","sort":{"field":"load_kg_max","direction":"desc"}}' --limit 20 --output-shape aggregate
  kura read_query --source-id performance_tests --read-kind timeline --subject test_timeline --filters-json '{"test_type":"cmj","measured_property":"jump_height_cm","date_from":"2026-01-01"}' --output-shape series

Rules:
  Use read_query for facts and bounded summaries. Use analyze for trend, stagnation, driver, influence, or training-decision questions.
  Use source_id=performance_tests for benchmark, jump-test, and performance-test history questions; common aliases such as benchmarks are normalized to performance_tests.
  Use activity_session_history day_history for compact multi-day recall, session_detail for one exact saved session, and block_history plus --query-json for programmable activity history reads.
  Use activity_session_history comparison_stream for exact progress; exercise_identity is only the movement-family view when variants are mixed.
  Continue with the returned cursor only when continuation.has_more=true.
"#;

pub fn public_read_query_payload_contract() -> Value {
    json!({
        "schema_version": "public_read_query_payload_contract.v1",
        "entrypoint": "kura read_query",
        "truth_namespace": "canonical_read_query",
        "principles": [
            "Use exactly one canonical source_id per call.",
            "Use cursor only for the continuation returned by a previous read_query response.",
            "For activity_session_history, use day_history for compact recall, session_detail for exact session detail, and block_history plus query for bounded programmable history reads.",
            "For activity_session_history, use comparison_stream as the exact progress grouping; exercise_identity is only the movement-family view when variants are mixed."
        ],
        "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": "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": "exercise_identity",
                        "aggregate": "load_kg_max",
                        "sort": {
                            "field": "load_kg_max",
                            "direction": "desc"
                        }
                    },
                    "limit": 20,
                    "output_shape": "aggregate"
                }
            },
            {
                "label": "Performance-test 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"
                }
            }
        ]
    })
}

#[derive(Subcommand)]
pub enum ReadCommands {
    /// List the machine-readable readable-source catalog or one source capability card
    Sources(ReadSourcesArgs),
    /// Run one declarative read_query request against the universal public read surface
    Query(ReadQueryArgs),
}

#[derive(Args)]
pub struct ReadSourcesArgs {
    /// Optional source id to return only one source capability card
    #[arg(long, alias = "source")]
    source_id: Option<String>,
}

#[derive(Args)]
#[command(after_help = READ_QUERY_HELP)]
pub struct ReadQueryArgs {
    /// Readable source id for one canonical read_query call
    #[arg(long)]
    source_id: String,
    /// Retrieval shape for this read
    #[arg(long, value_enum)]
    read_kind: ReadKindArg,
    /// Optional source-specific subject
    #[arg(long)]
    subject: Option<String>,
    /// Exact lookup key for projection-style sources
    #[arg(long)]
    lookup_key: Option<String>,
    /// Absolute local date in YYYY-MM-DD
    #[arg(long)]
    date: Option<String>,
    /// Absolute local start date in YYYY-MM-DD
    #[arg(long)]
    date_from: Option<String>,
    /// Absolute local end date in YYYY-MM-DD
    #[arg(long)]
    date_to: Option<String>,
    /// Full source-specific filters JSON object
    #[arg(long = "filters-json", alias = "filters")]
    filters_json: Option<String>,
    /// Optional bounded query JSON object for compact programmable reads
    #[arg(long = "query-json", alias = "query")]
    query_json: Option<String>,
    /// Optional exact page size
    #[arg(long)]
    limit: Option<u32>,
    /// Optional opaque continuation cursor
    #[arg(long)]
    cursor: Option<String>,
    /// Optional server-owned result handle
    #[arg(long)]
    result_handle: Option<String>,
    /// Optional response shape override
    #[arg(long, value_enum)]
    output_shape: Option<ReadOutputShapeArg>,
    /// Optional analysis escalation mode
    #[arg(long, value_enum)]
    analysis_mode: Option<ReadAnalysisModeArg>,
}

#[derive(Copy, Clone, Debug, Eq, PartialEq, ValueEnum)]
pub enum ReadKindArg {
    Lookup,
    List,
    Timeline,
    Aggregate,
}

impl ReadKindArg {
    fn as_contract_str(self) -> &'static str {
        match self {
            Self::Lookup => "lookup",
            Self::List => "list",
            Self::Timeline => "timeline",
            Self::Aggregate => "aggregate",
        }
    }
}

#[derive(Copy, Clone, Debug, Eq, PartialEq, ValueEnum)]
pub enum ReadOutputShapeArg {
    Item,
    Collection,
    Series,
    Aggregate,
    #[value(name = "analysis_handoff")]
    AnalysisHandoff,
}

impl ReadOutputShapeArg {
    fn as_contract_str(self) -> &'static str {
        match self {
            Self::Item => "item",
            Self::Collection => "collection",
            Self::Series => "series",
            Self::Aggregate => "aggregate",
            Self::AnalysisHandoff => "analysis_handoff",
        }
    }
}

#[derive(Copy, Clone, Debug, Eq, PartialEq, ValueEnum)]
pub enum ReadAnalysisModeArg {
    None,
    #[value(name = "allow_hybrid")]
    AllowHybrid,
    #[value(name = "require_analysis")]
    RequireAnalysis,
}

impl ReadAnalysisModeArg {
    fn as_contract_str(self) -> &'static str {
        match self {
            Self::None => "none",
            Self::AllowHybrid => "allow_hybrid",
            Self::RequireAnalysis => "require_analysis",
        }
    }
}

pub async fn run(api_url: &str, token: Option<&str>, command: ReadCommands) -> i32 {
    match command {
        ReadCommands::Sources(args) => read_sources(api_url, token, args).await,
        ReadCommands::Query(args) => read_query(api_url, token, args).await,
    }
}

pub async fn run_query(api_url: &str, token: Option<&str>, args: ReadQueryArgs) -> i32 {
    read_query(api_url, token, args).await
}

async fn read_sources(api_url: &str, token: Option<&str>, args: ReadSourcesArgs) -> i32 {
    let query = vec![(
        "section".to_string(),
        READ_SOURCE_CATALOG_SECTION.to_string(),
    )];

    if dry_run_enabled() {
        return emit_dry_run_request(
            &reqwest::Method::GET,
            api_url,
            READ_SOURCE_CATALOG_ENDPOINT,
            token.is_some(),
            None,
            &query,
            &[],
            false,
            Some("Fetch the machine-readable read source catalog for agent routing."),
        );
    }

    let response = raw_api_request_with_query(
        api_url,
        reqwest::Method::GET,
        READ_SOURCE_CATALOG_ENDPOINT,
        token,
        &query,
    )
    .await
    .unwrap_or_else(|error| {
        exit_error(
            &format!("Failed to reach {READ_SOURCE_CATALOG_ENDPOINT}: {error}"),
            Some("Check API availability/auth and retry."),
        )
    });

    let (status, body) = response;
    if (200..=299).contains(&status) {
        let output = build_read_sources_output(body, args.source_id.as_deref())
            .unwrap_or_else(|(message, docs_hint)| exit_error(&message, docs_hint.as_deref()));
        print_json_stdout(&output);
        0
    } else {
        print_json_stderr(&body);
        if (400..=499).contains(&status) { 1 } else { 2 }
    }
}

async fn read_query(api_url: &str, token: Option<&str>, args: ReadQueryArgs) -> i32 {
    let body = build_read_query_body(&args);
    execute_json_read(api_url, token, READ_QUERY_ENDPOINT, body).await
}

async fn execute_json_read(api_url: &str, token: Option<&str>, path: &str, body: Value) -> i32 {
    if dry_run_enabled() {
        return emit_dry_run_request(
            &reqwest::Method::POST,
            api_url,
            path,
            token.is_some(),
            Some(&body),
            &[],
            &[],
            false,
            None,
        );
    }

    let response = raw_api_request_json(
        api_url,
        reqwest::Method::POST,
        path,
        token,
        Some(body),
        &[],
        &[],
    )
    .await
    .unwrap_or_else(|error| {
        exit_error(
            &format!("Failed to reach {path}: {error}"),
            Some("Check API availability/auth and retry."),
        )
    });

    emit_json_response(response)
}

fn emit_json_response(response: RawApiResponse) -> i32 {
    if (200..=299).contains(&response.status) {
        print_json_stdout(&response.body);
        0
    } else {
        print_json_stderr(&response.body);
        if (400..=499).contains(&response.status) {
            1
        } else {
            2
        }
    }
}

fn build_read_query_body(args: &ReadQueryArgs) -> Value {
    let source_id =
        canonical_read_source_id(&required_non_empty_string(&args.source_id, "--source-id"));
    if matches!(
        source_id.as_str(),
        "authoritative_training_memory" | "training_timeline"
    ) {
        exit_error(
            "`kura read_query` no longer accepts removed legacy training sources.",
            Some(
                "Use --source-id activity_session_history for saved training/activity history, or use `kura analyze` for trend and driver questions.",
            ),
        );
    }

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

    if let Some(subject) = optional_non_empty_string(args.subject.as_deref()) {
        body.insert("subject".to_string(), json!(subject));
    }

    let mut filters =
        parse_optional_json_object_arg(args.filters_json.as_deref(), "--filters-json");
    insert_optional_string_checked(
        &mut filters,
        "lookup_key",
        args.lookup_key.as_deref(),
        "--lookup-key",
    );
    insert_optional_string_checked(&mut filters, "date", args.date.as_deref(), "--date");
    insert_optional_string_checked(
        &mut filters,
        "date_from",
        args.date_from.as_deref(),
        "--date-from",
    );
    insert_optional_string_checked(
        &mut filters,
        "date_to",
        args.date_to.as_deref(),
        "--date-to",
    );
    if !filters.is_empty() {
        body.insert("filters".to_string(), Value::Object(filters));
    }
    let query = parse_optional_json_object_arg(args.query_json.as_deref(), "--query-json");
    if !query.is_empty() {
        body.insert("query".to_string(), Value::Object(query));
    }

    if let Some(limit) = args.limit {
        body.insert("limit".to_string(), json!(limit));
    }
    insert_optional_string(&mut body, "cursor", args.cursor.as_deref());
    insert_optional_string(&mut body, "result_handle", args.result_handle.as_deref());
    if let Some(output_shape) = args.output_shape {
        body.insert(
            "output_shape".to_string(),
            json!(output_shape.as_contract_str()),
        );
    }
    if let Some(analysis_mode) = args.analysis_mode {
        body.insert(
            "analysis_mode".to_string(),
            json!(analysis_mode.as_contract_str()),
        );
    }

    Value::Object(body)
}

fn build_read_sources_output(
    section_body: Value,
    selected_source_id: Option<&str>,
) -> Result<Value, (String, Option<String>)> {
    let catalog = section_body
        .pointer("/value/contract")
        .and_then(Value::as_object)
        .cloned()
        .ok_or_else(|| {
            (
                "Read source catalog section did not expose a contract payload.".to_string(),
                Some(
                    "Check system_config.conventions::public_agent_read_source_catalog_v1 and retry."
                        .to_string(),
                ),
            )
        })?;
    let root = section_body.as_object().ok_or_else(|| {
        (
            "Read source catalog response must be a JSON object.".to_string(),
            Some("Retry once the server returns the readable source catalog section.".to_string()),
        )
    })?;

    let catalog_ref = READ_SOURCE_CATALOG_SECTION
        .split("::")
        .last()
        .unwrap_or("public_agent_read_source_catalog_v1");

    let section = root
        .get("section")
        .and_then(Value::as_str)
        .unwrap_or(READ_SOURCE_CATALOG_SECTION);
    let handle = root.get("handle").cloned().unwrap_or(Value::Null);
    let version = root.get("version").cloned().unwrap_or(Value::Null);
    let updated_at = root.get("updated_at").cloned().unwrap_or(Value::Null);

    if let Some(source_id) =
        selected_source_id.and_then(|value| optional_non_empty_string(Some(value)))
    {
        let sources = catalog
            .get("sources")
            .and_then(Value::as_object)
            .ok_or_else(|| {
                (
                    "Read source catalog contract did not expose a sources map.".to_string(),
                    Some("Check the source catalog contract and retry.".to_string()),
                )
            })?;
        let source = sources.get(source_id.as_str()).cloned().ok_or_else(|| {
            let available = sources.keys().cloned().collect::<Vec<_>>();
            (
                format!("Unknown read source_id '{source_id}'."),
                Some(format!(
                    "Use one declared source_id from the read_query schema, for example: {}.",
                    available.join(", ")
                )),
            )
        })?;
        return Ok(json!({
            "catalog_ref": catalog_ref,
            "catalog_section": section,
            "catalog_handle": handle,
            "catalog_version": version,
            "catalog_updated_at": updated_at,
            "source_id": source_id,
            "source": source,
        }));
    }

    let mut output = catalog;
    output.insert("catalog_ref".to_string(), json!(catalog_ref));
    output.insert("catalog_section".to_string(), json!(section));
    output.insert("catalog_handle".to_string(), handle);
    output.insert("catalog_version".to_string(), version);
    output.insert("catalog_updated_at".to_string(), updated_at);
    Ok(Value::Object(output))
}

fn optional_non_empty_string(raw: Option<&str>) -> Option<String> {
    raw.map(str::trim)
        .filter(|value| !value.is_empty())
        .map(str::to_string)
}

fn required_non_empty_string(raw: &str, flag_name: &str) -> String {
    optional_non_empty_string(Some(raw)).unwrap_or_else(|| {
        exit_error(
            &format!("`kura read` requires a non-empty {flag_name} value."),
            Some("Provide the exact contract value instead of an empty string."),
        )
    })
}

fn canonical_read_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(),
        _ => source_id.trim().to_string(),
    }
}

fn parse_optional_json_object_arg(raw: Option<&str>, flag_name: &str) -> Map<String, Value> {
    let Some(raw) = optional_non_empty_string(raw) else {
        return Map::new();
    };
    let parsed: Value = serde_json::from_str(raw.as_str()).unwrap_or_else(|error| {
        exit_error(
            &format!("`kura read_query` could not parse {flag_name} as JSON: {error}"),
            Some("Pass a JSON object, for example --filters-json '{\"date\":\"2026-04-02\"}'."),
        )
    });
    parsed.as_object().cloned().unwrap_or_else(|| {
        exit_error(
            &format!("`kura read_query` requires {flag_name} to be a JSON object."),
            Some("Use an object like '{\"date_from\":\"2026-04-01\"}', not an array or string."),
        )
    })
}

fn insert_json_field_checked(
    target: &mut Map<String, Value>,
    field: &str,
    value: Value,
    flag_name: &str,
) {
    if let Some(existing) = target.get(field) {
        if existing == &value {
            return;
        }
        exit_error(
            &format!("`kura read_query` received {field} twice with different values."),
            Some(&format!(
                "Use either {flag_name} or the same field inside the JSON object, not both."
            )),
        );
    }
    target.insert(field.to_string(), value);
}

fn insert_optional_string_checked(
    target: &mut Map<String, Value>,
    field: &str,
    raw: Option<&str>,
    flag_name: &str,
) {
    if let Some(value) = optional_non_empty_string(raw) {
        insert_json_field_checked(target, field, json!(value), flag_name);
    }
}

fn insert_optional_string(target: &mut Map<String, Value>, field: &str, raw: Option<&str>) {
    if let Some(value) = optional_non_empty_string(raw) {
        target.insert(field.to_string(), json!(value));
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn build_read_query_body_serializes_live_public_read_contract() {
        let body = build_read_query_body(&ReadQueryArgs {
            source_id: "activity_session_history".to_string(),
            read_kind: ReadKindArg::Aggregate,
            subject: Some("activity_summary".to_string()),
            lookup_key: None,
            date: None,
            date_from: Some("2026-04-01".to_string()),
            date_to: Some("2026-04-07".to_string()),
            filters_json: None,
            query_json: None,
            limit: Some(20),
            cursor: Some("rq1:opaque".to_string()),
            result_handle: None,
            output_shape: Some(ReadOutputShapeArg::Aggregate),
            analysis_mode: Some(ReadAnalysisModeArg::AllowHybrid),
        });

        assert_eq!(body["schema_version"], json!(READ_QUERY_SCHEMA_VERSION));
        assert_eq!(body["source_id"], json!("activity_session_history"));
        assert_eq!(body["read_kind"], json!("aggregate"));
        assert_eq!(body["subject"], json!("activity_summary"));
        assert_eq!(body["filters"]["date_from"], json!("2026-04-01"));
        assert_eq!(body["filters"]["date_to"], json!("2026-04-07"));
        assert_eq!(body["limit"], json!(20));
        assert_eq!(body["cursor"], json!("rq1:opaque"));
        assert_eq!(body["output_shape"], json!("aggregate"));
        assert_eq!(body["analysis_mode"], json!("allow_hybrid"));
    }

    #[test]
    fn build_read_query_body_omits_empty_optional_sections() {
        let body = build_read_query_body(&ReadQueryArgs {
            source_id: "recovery".to_string(),
            read_kind: ReadKindArg::Timeline,
            subject: None,
            lookup_key: Some("  ".to_string()),
            date: None,
            date_from: None,
            date_to: None,
            filters_json: None,
            query_json: None,
            limit: None,
            cursor: None,
            result_handle: None,
            output_shape: Some(ReadOutputShapeArg::Series),
            analysis_mode: None,
        });

        assert!(body.get("filters").is_none());
        assert_eq!(body["source_id"], json!("recovery"));
        assert_eq!(body["read_kind"], json!("timeline"));
        assert_eq!(body["output_shape"], json!("series"));
    }

    #[test]
    fn build_read_query_body_accepts_json_filters() {
        let body = build_read_query_body(&ReadQueryArgs {
            source_id: "activity_session_history".to_string(),
            read_kind: ReadKindArg::Aggregate,
            subject: None,
            lookup_key: None,
            date: None,
            date_from: None,
            date_to: None,
            filters_json: Some(r#"{"date_from":"2026-03-01","date_to":"2026-04-01","block_kind":"repetition_sets"}"#.to_string()),
            query_json: Some(r#"{"group_by":"exercise_identity","aggregate":"load_kg_max","sort":{"field":"load_kg_max","direction":"desc"}}"#.to_string()),
            limit: Some(10),
            cursor: None,
            result_handle: None,
            output_shape: Some(ReadOutputShapeArg::Aggregate),
            analysis_mode: Some(ReadAnalysisModeArg::AllowHybrid),
        });

        assert_eq!(body["filters"]["date_from"], json!("2026-03-01"));
        assert_eq!(body["filters"]["date_to"], json!("2026-04-01"));
        assert_eq!(body["filters"]["block_kind"], json!("repetition_sets"));
        assert_eq!(body["query"]["group_by"], json!("exercise_identity"));
        assert_eq!(body["query"]["aggregate"], json!("load_kg_max"));
    }

    #[test]
    fn build_read_query_body_normalizes_benchmark_source_aliases() {
        let body = build_read_query_body(&ReadQueryArgs {
            source_id: "benchmarks".to_string(),
            read_kind: ReadKindArg::Timeline,
            subject: Some("test_timeline".to_string()),
            lookup_key: None,
            date: None,
            date_from: None,
            date_to: None,
            filters_json: Some(r#"{"test_type":"cmj"}"#.to_string()),
            query_json: None,
            limit: None,
            cursor: None,
            result_handle: None,
            output_shape: Some(ReadOutputShapeArg::Series),
            analysis_mode: None,
        });

        assert_eq!(body["source_id"], json!("performance_tests"));
        assert_eq!(body["filters"]["test_type"], json!("cmj"));
    }

    #[test]
    fn build_read_sources_output_returns_single_source_card_when_requested() {
        let output = build_read_sources_output(
            json!({
                "section": READ_SOURCE_CATALOG_SECTION,
                "handle": "system_config/global@v9",
                "version": 9,
                "updated_at": "2026-04-02T10:00:00Z",
                "value": {
                    "rules": ["x"],
                    "contract": {
                        "schema_version": "public_agent_read_source_catalog.v1",
                        "sources": {
                            "recovery": {
                                "source_id": "recovery",
                                "supported_read_kinds": ["lookup", "list", "timeline", "aggregate"]
                            }
                        }
                    }
                }
            }),
            Some("recovery"),
        )
        .expect("single-source output should build");

        assert_eq!(
            output["catalog_ref"],
            json!("public_agent_read_source_catalog_v1")
        );
        assert_eq!(output["source_id"], json!("recovery"));
        assert_eq!(output["source"]["source_id"], json!("recovery"));
        assert_eq!(output["catalog_handle"], json!("system_config/global@v9"));
    }
}