#![allow(
    clippy::expect_used,
    clippy::unwrap_used,
    clippy::panic,
    clippy::uninlined_format_args,
    clippy::collapsible_if,
    clippy::redundant_clone,
    clippy::needless_raw_string_hashes,
    clippy::single_match,
    clippy::redundant_closure_for_method_calls,
    clippy::redundant_pattern_matching,
    clippy::ignored_unit_patterns,
    clippy::clone_on_copy,
    clippy::manual_assert,
    clippy::unwrap_in_result,
    clippy::useless_vec
)]
use std::process::Command;
use std::time::Duration;

use meerkat_mobkit::{
    AuthPolicy, AuthProvider, BigQueryNaming, ConsoleAccessRequest, ConsolePolicy,
    ConsoleRestJsonRequest, DiscoverySpec, JsonRpcResponse, MOBKIT_CONTRACT_VERSION, MobKitConfig,
    ModuleConfig, ModuleRouteRequest, PreSpawnData, RestartPolicy, RuntimeDecisionInputs,
    RuntimeOpsPolicy, TrustedOidcRuntimeConfig, UnifiedEvent, build_runtime_decision_state,
    handle_console_rest_json_route, handle_mobkit_rpc_json, route_module_call,
    run_discovered_module_once, start_mobkit_runtime,
};
use tempfile::TempDir;

fn shell_module(id: &str, script: &str) -> ModuleConfig {
    ModuleConfig {
        id: id.to_string(),
        command: "sh".to_string(),
        args: vec!["-c".to_string(), script.to_string()],
        restart_policy: RestartPolicy::Never,
    }
}

fn release_json() -> String {
    include_str!("../assets/release-targets.json").to_string()
}

fn trusted_oidc() -> TrustedOidcRuntimeConfig {
    TrustedOidcRuntimeConfig {
        discovery_json:
            r#"{"issuer":"https://trusted.mobkit.local","jwks_uri":"https://trusted.mobkit.local/.well-known/jwks.json"}"#
                .to_string(),
        jwks_json: r#"{"keys":[{"kid":"kid-current","kty":"oct","alg":"HS256","k":"cGhhc2U3LXRydXN0ZWQtY3VycmVudC1zZWNyZXQ"},{"kid":"kid-next","kty":"oct","alg":"HS256","k":"cGhhc2U3LXRydXN0ZWQtbmV4dC1zZWNyZXQ"}]}"#.to_string(),
        audience: "meerkat-console".to_string(),
    }
}

fn trusted_toml_for_sdk_console_probe() -> String {
    r#"
[[modules]]
id = "scheduling"
command = "scheduler-bin"
args = ["--poll-ms", "250"]
restart_policy = "on_failure"

[[modules]]
id = "routing"
command = "router-bin"
args = ["--mode", "fast"]
restart_policy = "always"
"#
    .to_string()
}

fn module_probe_config(module_ids: &[&str]) -> MobKitConfig {
    let modules = module_ids
        .iter()
        .enumerate()
        .map(|(idx, module_id)| {
            let tool_list = format!("{module_id}/tools.list");
            let representative_call = format!("{module_id}/tool.call");
            let envelope = serde_json::json!({
                "event_id": format!("evt-{module_id}"),
                "source": "module",
                "timestamp_ms": 100 + idx as u64,
                "event": {
                    "kind": "module",
                    "module": module_id,
                    "event_type": "ready",
                    "payload": {
                        "family": module_id,
                        "health": {"state": "healthy"},
                        "tools": {
                            "list_method": tool_list,
                            "representative_call": {
                                "method": representative_call,
                                "params_schema": {"tool": "string", "input": "json"}
                            }
                        }
                    }
                }
            });
            shell_module(module_id, &format!("printf '%s\\n' '{}'", envelope))
        })
        .collect();

    let discovery_modules = module_ids
        .iter()
        .map(|module| (*module).to_string())
        .collect();
    let pre_spawn = module_ids
        .iter()
        .map(|module_id| PreSpawnData {
            module_id: (*module_id).to_string(),
            env: vec![("MODULE_FAMILY".to_string(), (*module_id).to_string())],
        })
        .collect();

    MobKitConfig {
        modules,
        discovery: DiscoverySpec {
            namespace: "phase0b".to_string(),
            modules: discovery_modules,
        },
        pre_spawn,
    }
}

fn run_shell(script: &str, env: &[(&str, &str)]) -> String {
    let mut command = Command::new("sh");
    command.arg("-c").arg(script);
    for (k, v) in env {
        command.env(k, v);
    }
    let output = command.output().expect("shell command should start");
    assert!(
        output.status.success(),
        "command failed: {}\nstdout:\n{}\nstderr:\n{}",
        script,
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );
    String::from_utf8(output.stdout).expect("stdout utf8")
}

fn gate_enabled(flag: &str) -> bool {
    std::env::var(flag).ok().as_deref() == Some("1")
}

fn skip_unless_gate_enabled(flag: &str) -> bool {
    if gate_enabled(flag) {
        true
    } else {
        eprintln!("skipping external boundary test because {flag} is not set to 1");
        false
    }
}

const DEFAULT_P0B_BQ_PROJECT: &str = "king-dnn-training-dev";

fn run_phase0b_rpc_gateway(request_json: &str) -> String {
    let output = Command::new(env!("CARGO_BIN_EXE_rpc_gateway"))
        .env("MOBKIT_RPC_REQUEST", request_json)
        .output()
        .expect("phase0b rpc gateway should start");
    assert!(
        output.status.success(),
        "phase0b_rpc_gateway failed\nstdout:\n{}\nstderr:\n{}",
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );
    String::from_utf8(output.stdout).expect("stdout utf8")
}

fn module_family_process_config(module_ids: &[&str]) -> (MobKitConfig, TempDir) {
    let temp = tempfile::tempdir().expect("tempdir for module-family probes");
    let mut modules = Vec::new();
    let mut discovery_modules = Vec::new();
    let mut pre_spawn = Vec::new();

    for (idx, module_id) in module_ids.iter().enumerate() {
        let script_path = temp.path().join(format!("probe_{module_id}.py"));
        let script = format!(
            r#"#!/usr/bin/env python3
import json
import os

family = "{module_id}"
if os.environ.get("MODULE_FAMILY") != family:
    raise SystemExit(17)
envelope = {{
    "event_id": "evt-{module_id}-{idx}",
    "source": "module",
    "timestamp_ms": {timestamp_ms},
    "event": {{
        "kind": "module",
        "module": family,
        "event_type": "ready",
        "payload": {{
            "family": family,
            "health": {{"state": "healthy"}},
            "process_probe": "python-script-{module_id}",
            "tools": {{
                "list_method": f"{{family}}/tools.list",
                "representative_call": {{
                    "method": f"{{family}}/tool.call",
                    "params_schema": {{"tool": "string", "input": "json"}}
                }}
            }}
        }}
    }}
}}
print(json.dumps(envelope))
"#,
            timestamp_ms = 1_000 + idx as u64
        );
        std::fs::write(&script_path, script).expect("write module-family probe script");

        modules.push(ModuleConfig {
            id: (*module_id).to_string(),
            command: "python3".to_string(),
            args: vec![script_path.to_string_lossy().to_string()],
            restart_policy: RestartPolicy::Never,
        });
        discovery_modules.push((*module_id).to_string());
        pre_spawn.push(PreSpawnData {
            module_id: (*module_id).to_string(),
            env: vec![("MODULE_FAMILY".to_string(), (*module_id).to_string())],
        });
    }

    (
        MobKitConfig {
            modules,
            discovery: DiscoverySpec {
                namespace: "phase0b-family-process".to_string(),
                modules: discovery_modules,
            },
            pre_spawn,
        },
        temp,
    )
}

#[test]
#[ignore = "external roadmap preflight check (phase 0b)"]
fn p0b_t1_rpc_method_matrix_boundary_preflight() {
    let status: JsonRpcResponse = serde_json::from_str(&run_phase0b_rpc_gateway(
        r#"{"jsonrpc":"2.0","id":"1","method":"mobkit/status","params":{}}"#,
    ))
    .expect("status response json");
    assert_eq!(status.id, serde_json::json!("1"));
    assert_eq!(
        status
            .result
            .as_ref()
            .and_then(|result| result.get("contract_version"))
            .and_then(|value| value.as_str()),
        Some(MOBKIT_CONTRACT_VERSION)
    );
    assert_eq!(
        status
            .result
            .as_ref()
            .and_then(|result| result.get("loaded_modules"))
            .and_then(|value| value.as_array())
            .map(|modules| modules.len()),
        Some(1)
    );

    let capabilities: JsonRpcResponse = serde_json::from_str(&run_phase0b_rpc_gateway(
        r#"{"jsonrpc":"2.0","id":"2","method":"mobkit/capabilities","params":{}}"#,
    ))
    .expect("capabilities response json");
    assert_eq!(capabilities.id, serde_json::json!("2"));
    let methods = capabilities
        .result
        .as_ref()
        .and_then(|result| result.get("methods"))
        .and_then(|value| value.as_array())
        .expect("methods array");
    assert!(methods.iter().any(|method| method == "mobkit/status"));

    let routed: JsonRpcResponse = serde_json::from_str(&run_phase0b_rpc_gateway(
        r#"{"jsonrpc":"2.0","id":"3","method":"routing/tools.list","params":{"probe":"viability"}}"#,
    ))
    .expect("module proxy response json");
    assert_eq!(
        routed
            .result
            .as_ref()
            .and_then(|result| result.get("module_id"))
            .and_then(|value| value.as_str()),
        Some("routing")
    );

    let not_found: JsonRpcResponse = serde_json::from_str(&run_phase0b_rpc_gateway(
        r#"{"jsonrpc":"2.0","id":"4","method":"mobkit/not_loaded","params":{}}"#,
    ))
    .expect("unknown method response json");
    assert_eq!(not_found.error.map(|err| err.code), Some(-32601));

    let unloaded: JsonRpcResponse = serde_json::from_str(&run_phase0b_rpc_gateway(
        r#"{"jsonrpc":"2.0","id":"5","method":"delivery/tools.list","params":{"probe":"viability"}}"#,
    ))
    .expect("unloaded route response json");
    assert_eq!(unloaded.error.map(|err| err.code), Some(-32601));

    let notification =
        run_phase0b_rpc_gateway(r#"{"jsonrpc":"2.0","method":"mobkit/status","params":{}}"#);
    assert!(notification.is_empty());
}

#[test]
#[ignore = "external roadmap preflight check (phase 0b)"]
fn p0b_t2_bigquery_real_boundary_on_king_dnn_training_dev() {
    if !skip_unless_gate_enabled("MOBKIT_RUN_P0B_BQ") {
        return;
    }
    let project = std::env::var("MOBKIT_P0B_BQ_PROJECT")
        .ok()
        .filter(|value| !value.trim().is_empty())
        .unwrap_or_else(|| DEFAULT_P0B_BQ_PROJECT.to_string());
    let stdout = run_shell(
        r#"
set -euo pipefail
PROJECT="${MOBKIT_P0B_BQ_PROJECT:-king-dnn-training-dev}"
echo "PROJECT_SELECTED:$(gcloud projects describe "$PROJECT" --format='value(projectId)')"
UNIQ="$(python3 - <<'PY'
import uuid
print(uuid.uuid4().hex)
PY
)"
DATASET="mobkit_p0b_${UNIQ}"
TABLE="sessions"
STREAM_TABLE="sessions_stream_probe"
cleanup() {
  set +e
  echo "CLEANUP_BEGIN:${PROJECT}:${DATASET}.${TABLE},${STREAM_TABLE}"
  bq --project_id="$PROJECT" rm -f -t "${PROJECT}:${DATASET}.${TABLE}" >/dev/null 2>&1 || true
  bq --project_id="$PROJECT" rm -f -t "${PROJECT}:${DATASET}.${STREAM_TABLE}" >/dev/null 2>&1 || true
  bq --project_id="$PROJECT" rm -f -d "${PROJECT}:${DATASET}" >/dev/null 2>&1 || true
  echo "CLEANUP_END:${PROJECT}:${DATASET}"
}
trap cleanup EXIT
bq --project_id="$PROJECT" --location=US mk -d --description "mobkit phase0b preflight" "${PROJECT}:${DATASET}"
echo "DATASET_CREATED:${PROJECT}:${DATASET}"
bq --project_id="$PROJECT" --location=US mk \
  --table "${PROJECT}:${DATASET}.${TABLE}" \
  session_id:STRING,updated_at:TIMESTAMP,deleted:BOOL,payload:STRING
echo "TABLE_CREATED:${PROJECT}:${DATASET}.${TABLE}"
bq --project_id="$PROJECT" --location=US mk \
  --table "${PROJECT}:${DATASET}.${STREAM_TABLE}" \
  session_id:STRING,updated_at:TIMESTAMP,deleted:BOOL,payload:STRING
ROWS_FILE="${DATASET}_stream_rows.ndjson"
cat > "${ROWS_FILE}" <<'EOF'
{"session_id":"stream-s1","updated_at":"2024-01-01T00:00:00Z","deleted":false,"payload":"{\"stream\":true}"}
EOF
bq --project_id="$PROJECT" insert "${PROJECT}:${DATASET}.${STREAM_TABLE}" "${ROWS_FILE}"
rm -f "${ROWS_FILE}"
for _ in $(seq 1 60); do
  STREAM_PROBE_COUNT="$(bq --project_id="$PROJECT" query --nouse_legacy_sql --format=csv \
    "SELECT COUNT(*) AS c FROM \`${PROJECT}.${DATASET}.${STREAM_TABLE}\`" | tail -n +2 | tr -d '\r')"
  [ "${STREAM_PROBE_COUNT}" = "1" ] && break
  sleep 1
done
echo "STREAM_PROBE_COUNT:${STREAM_PROBE_COUNT}"
[ "${STREAM_PROBE_COUNT}" = "1" ]
bq --project_id="$PROJECT" query --nouse_legacy_sql \
  "INSERT \`${PROJECT}.${DATASET}.${TABLE}\` (session_id,updated_at,deleted,payload) VALUES ('s1', TIMESTAMP('2024-01-01T00:00:00Z'), FALSE, '{\"ok\":true}')"
bq --project_id="$PROJECT" query --nouse_legacy_sql \
  "INSERT \`${PROJECT}.${DATASET}.${TABLE}\` (session_id,updated_at,deleted,payload) VALUES ('s1', TIMESTAMP('2024-01-01T00:00:01Z'), TRUE, '')"
bq --project_id="$PROJECT" query --nouse_legacy_sql \
  "INSERT \`${PROJECT}.${DATASET}.${TABLE}\` (session_id,updated_at,deleted,payload) VALUES ('s2', TIMESTAMP('2024-01-01T00:00:00Z'), FALSE, '{\"ok\":false}')"
DEDUP_CSV="$(bq --project_id="$PROJECT" query --nouse_legacy_sql --format=csv \
  "SELECT session_id, deleted FROM \`${PROJECT}.${DATASET}.${TABLE}\` QUALIFY ROW_NUMBER() OVER (PARTITION BY session_id ORDER BY updated_at DESC)=1 ORDER BY session_id")"
DEDUP_ROWS="$(echo "$DEDUP_CSV" | tail -n +2 | tr -d '\r')"
while IFS= read -r row; do
  [ -n "$row" ] && echo "DEDUP_ROW:${row}"
done <<EOF
$DEDUP_ROWS
EOF
TOMBSTONE_COUNT="$(bq --project_id="$PROJECT" query --nouse_legacy_sql --format=csv \
  "SELECT COUNT(*) AS c FROM \`${PROJECT}.${DATASET}.${TABLE}\` WHERE session_id='s1' AND deleted=TRUE" | tail -n +2 | tr -d '\r')"
echo "TOMBSTONE_COUNT:${TOMBSTONE_COUNT}"
bq --project_id="$PROJECT" query --nouse_legacy_sql "TRUNCATE TABLE \`${PROJECT}.${DATASET}.${TABLE}\`" >/dev/null
POST_TRUNCATE_COUNT="$(bq --project_id="$PROJECT" query --nouse_legacy_sql --format=csv \
  "SELECT COUNT(*) AS c FROM \`${PROJECT}.${DATASET}.${TABLE}\`" | tail -n +2 | tr -d '\r')"
echo "POST_TRUNCATE_COUNT:${POST_TRUNCATE_COUNT}"
"#,
        &[("MOBKIT_P0B_BQ_PROJECT", project.as_str())],
    );
    println!("{stdout}");

    assert!(stdout.contains(&format!("PROJECT_SELECTED:{project}")));
    assert!(stdout.contains(&format!("DATASET_CREATED:{project}:mobkit_p0b_")));
    assert!(stdout.contains(&format!("TABLE_CREATED:{project}:mobkit_p0b_")));
    assert!(stdout.contains("STREAM_PROBE_COUNT:1"));
    assert!(stdout.contains("DEDUP_ROW:s1,true"));
    assert!(stdout.contains("DEDUP_ROW:s2,false"));
    assert!(stdout.contains("TOMBSTONE_COUNT:1"));
    assert!(stdout.contains("POST_TRUNCATE_COUNT:0"));
    assert!(stdout.contains(&format!("CLEANUP_BEGIN:{project}:mobkit_p0b_")));
    assert!(stdout.contains(&format!("CLEANUP_END:{project}:mobkit_p0b_")));
}

#[test]
#[ignore = "external roadmap preflight check (phase 0b)"]
fn p0b_t3_json_store_filesystem_lock_preflight() {
    let output = run_shell(
        r#"
set -euo pipefail
TMPDIR="$(mktemp -d)"
LOCKFILE="${TMPDIR}/lockfile"
READYFILE="${TMPDIR}/holder_ready"
cleanup() {
  set +e
  if [ -n "${HOLDER_PID:-}" ]; then
    kill "${HOLDER_PID}" >/dev/null 2>&1 || true
    wait "${HOLDER_PID}" 2>/dev/null || true
  fi
  rm -rf "${TMPDIR}"
}
trap cleanup EXIT
python3 - <<'PY' "$LOCKFILE" "$READYFILE" &
import fcntl, sys, time
lockfile, readyfile = sys.argv[1], sys.argv[2]
f = open(lockfile, "w")
fcntl.flock(f, fcntl.LOCK_EX)
open(readyfile, "w").write("ready\n")
time.sleep(10)
PY
HOLDER_PID=$!
for _ in $(seq 1 100); do
  [ -f "$READYFILE" ] && break
  sleep 0.05
done
[ -f "$READYFILE" ]
HELD_RESULT="$(python3 -c 'import fcntl,sys; f=open(sys.argv[1],"w");
try:
    fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB); print("acquired")
except BlockingIOError:
    print("blocked")' "$LOCKFILE")"
echo "LOCK_WHILE_HELD:${HELD_RESULT}"
kill "${HOLDER_PID}" >/dev/null 2>&1 || true
wait "${HOLDER_PID}" 2>/dev/null || true
unset HOLDER_PID
RECOVERY_RESULT="$(python3 -c 'import fcntl,sys; f=open(sys.argv[1],"w");
try:
    fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB); print("acquired")
except BlockingIOError:
    print("blocked")' "$LOCKFILE")"
echo "LOCK_AFTER_RELEASE:${RECOVERY_RESULT}"
"#,
        &[],
    );
    assert!(output.contains("LOCK_WHILE_HELD:blocked"));
    assert!(output.contains("LOCK_AFTER_RELEASE:acquired"));
}

#[test]
#[ignore = "external roadmap preflight check (phase 0b)"]
fn p0b_t4_sse_stream_and_reconnect_preflight() {
    let output = run_shell(
        r#"
set -euo pipefail
TMPDIR="$(mktemp -d)"
cleanup() {
  set +e
  if [ -n "${PID:-}" ]; then
    kill "${PID}" >/dev/null 2>&1 || true
    wait "${PID}" 2>/dev/null || true
  fi
  rm -rf "${TMPDIR}"
}
trap cleanup EXIT
cat > "${TMPDIR}/sse_server.py" <<'PY'
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

class H(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/health":
            self.send_response(200)
            self.end_headers()
            self.wfile.write(b"ok")
            self.wfile.flush()
            return
        if self.path.startswith("/events"):
            last = self.headers.get("Last-Event-ID", "")
            self.send_response(200)
            self.send_header("Content-Type", "text/event-stream")
            self.send_header("Cache-Control", "no-cache")
            self.end_headers()
            if last == "2":
                self.wfile.write(b": keep-alive\n\n")
                self.wfile.write(b"id: 3\n")
                self.wfile.write(b"event: message\n")
                self.wfile.write(b"data: {\"kind\":\"backfill\",\"from\":\"2\"}\n\n")
            else:
                self.wfile.write(b"id: 1\n")
                self.wfile.write(b"event: message\n")
                self.wfile.write(b"data: {\"kind\":\"ready\"}\n\n")
                self.wfile.write(b"id: 2\n")
                self.wfile.write(b"event: message\n")
                self.wfile.write(b"data: {\"kind\":\"steady\"}\n\n")
            self.wfile.flush()
            return
        self.send_response(404)
        self.end_headers()
    def log_message(self, *_):
        return
server = ThreadingHTTPServer(("127.0.0.1", 0), H)
port = server.server_address[1]
print(port, flush=True)
server.serve_forever()
PY
python3 "${TMPDIR}/sse_server.py" > "${TMPDIR}/port.txt" &
PID=$!
for _ in $(seq 1 100); do
  [ -s "${TMPDIR}/port.txt" ] && break
  sleep 0.05
done
PORT="$(cat "${TMPDIR}/port.txt")"
for _ in $(seq 1 100); do
  curl -fsS --max-time 1 "http://127.0.0.1:${PORT}/health" >/dev/null && break
  sleep 0.05
done
curl -fsS --max-time 1 "http://127.0.0.1:${PORT}/health" >/dev/null
FIRST="$(curl -sN --max-time 2 "http://127.0.0.1:${PORT}/events")"
LAST_ID="$(printf '%s\n' "$FIRST" | awk '/^id: / {print $2}' | tail -n1)"
SECOND="$(curl -sN --max-time 2 -H "Last-Event-ID: ${LAST_ID}" "http://127.0.0.1:${PORT}/events")"
KEEPALIVE_COUNT="$(printf '%s\n' "$SECOND" | awk '/^: keep-alive$/ {c++} END {print c+0}')"
echo "FIRST_LAST_ID:${LAST_ID}"
echo "KEEPALIVE_COUNT:${KEEPALIVE_COUNT}"
printf '%s\n' "$FIRST" | awk '/^data: / {print}'
printf '%s\n' "$SECOND" | awk '/^data: / {print}'
"#,
        &[],
    );
    assert!(output.contains("FIRST_LAST_ID:2"));
    assert!(output.contains("KEEPALIVE_COUNT:1"));
    assert!(output.contains("data: {\"kind\":\"ready\"}"));
    assert!(output.contains("data: {\"kind\":\"steady\"}"));
    assert!(output.contains("data: {\"kind\":\"backfill\",\"from\":\"2\"}"));
}

#[test]
#[ignore = "external roadmap preflight check (phase 0b)"]
fn p0b_t5_auth_oidc_jwks_endpoint_preflight() {
    if !skip_unless_gate_enabled("MOBKIT_RUN_P0B_OIDC") {
        return;
    }

    let output = run_shell(
        r#"
set -euo pipefail
CONF="$(curl -sS --max-time 10 https://accounts.google.com/.well-known/openid-configuration)"
JWKS="$(CONF_JSON="$CONF" python3 -c 'import json,os; print(json.loads(os.environ["CONF_JSON"])["jwks_uri"])')"
echo "JWKS_URI:$JWKS"
KEY_COUNT="$(curl -sS -L --max-time 10 "$JWKS" | python3 -c 'import json,sys; print(len(json.loads(sys.stdin.read()).get("keys", [])))')"
echo "JWKS_KEY_COUNT:$KEY_COUNT"
set +e
CONF_JSON="$CONF" EXPECTED_JWKS="$JWKS" CANDIDATE_JWKS="${JWKS}?phase0b_mismatch=1" python3 - <<'PY'
import json
import os
import sys

discovered = json.loads(os.environ["CONF_JSON"])["jwks_uri"]
candidate = os.environ["CANDIDATE_JWKS"]
if discovered != os.environ["EXPECTED_JWKS"]:
    raise SystemExit(41)
if discovered != candidate:
    raise SystemExit(23)
raise SystemExit(0)
PY
MISMATCH_RC="$?"
set -e
echo "JWKS_MISMATCH_RC:${MISMATCH_RC}"
"#,
        &[],
    );
    let mut lines = output.lines();
    let jwks_uri = lines
        .next()
        .unwrap_or_default()
        .trim()
        .strip_prefix("JWKS_URI:")
        .unwrap_or_default()
        .to_string();
    let key_count = lines
        .next()
        .unwrap_or_default()
        .trim()
        .strip_prefix("JWKS_KEY_COUNT:")
        .unwrap_or_default()
        .to_string();
    let mismatch_rc = lines
        .next()
        .unwrap_or_default()
        .trim()
        .strip_prefix("JWKS_MISMATCH_RC:")
        .unwrap_or_default()
        .to_string();
    assert!(jwks_uri.starts_with("https://"));
    let count: usize = key_count.parse().expect("jwks key count");
    assert!(count > 0, "jwks must have at least one key");
    assert_eq!(mismatch_rc, "23");
}

#[test]
#[ignore = "external roadmap preflight check (phase 0b)"]
fn p0b_t6_module_family_process_preflight() {
    let module_families = ["scheduling", "routing", "delivery", "gating", "memory"];
    let (config, _temp_guard) = module_family_process_config(&module_families);

    for module_id in module_families {
        let event = run_discovered_module_once(&config, module_id, Duration::from_secs(1))
            .expect("module should emit ready event");
        assert_eq!(event.source, "module");
        match event.event {
            UnifiedEvent::Module(module_event) => {
                assert_eq!(module_event.module, module_id);
                assert_eq!(module_event.event_type, "ready");
                assert_eq!(module_event.payload["family"], module_id);
                assert_eq!(module_event.payload["health"]["state"], "healthy");
                assert_eq!(
                    module_event.payload["process_probe"],
                    format!("python-script-{module_id}")
                );
                assert_eq!(
                    module_event.payload["tools"]["list_method"],
                    format!("{module_id}/tools.list")
                );
                assert_eq!(
                    module_event.payload["tools"]["representative_call"]["method"],
                    format!("{module_id}/tool.call")
                );
            }
            other => panic!("expected module event, got {other:?}"),
        }
    }

    let runtime = start_mobkit_runtime(config, vec![], Duration::from_secs(1))
        .expect("runtime should start for module families");
    for module_id in module_families {
        let result = route_module_call(
            &runtime,
            &ModuleRouteRequest {
                module_id: module_id.to_string(),
                method: format!("{module_id}/tools.list"),
                params: serde_json::json!({"probe":"viability"}),
            },
            Duration::from_secs(1),
        );
        match module_id {
            "scheduling" | "delivery" | "memory" => {
                let err = result.expect_err("module should require MCP in runtime routing");
                let err_text = format!("{err:?}");
                assert!(
                    err_text.contains("McpRequired"),
                    "expected MCP-required routing error for {module_id}, got {err_text}"
                );
                assert!(
                    err_text.contains(module_id),
                    "expected module id in error for {module_id}, got {err_text}"
                );
            }
            _ => {
                let response = result.expect("module route should be viable");
                assert_eq!(response.module_id, module_id);
                assert_eq!(response.method, format!("{module_id}/tools.list"));
                assert_eq!(
                    response.payload["tools"]["representative_call"]["params_schema"]["tool"],
                    "string"
                );
            }
        }
    }
}

#[test]
#[ignore = "external roadmap preflight check (phase 0b)"]
fn p0b_t7_sdk_console_toolchain_and_payload_preflight() {
    let config = module_probe_config(&["scheduling", "routing"]);
    let mut runtime = start_mobkit_runtime(config, vec![], Duration::from_secs(1))
        .expect("runtime should start for sdk probe");

    let node_request = run_shell(
        r#"node -e 'const req={jsonrpc:"2.0",id:"sdk-node-1",method:"mobkit/capabilities",params:{}}; process.stdout.write(JSON.stringify(req));'"#,
        &[],
    );
    let rpc_response_json =
        handle_mobkit_rpc_json(&mut runtime, node_request.trim(), Duration::from_secs(1));
    let rpc_response: JsonRpcResponse =
        serde_json::from_str(&rpc_response_json).expect("rpc response should be valid");
    assert_eq!(rpc_response.id, serde_json::json!("sdk-node-1"));
    assert_eq!(
        rpc_response
            .result
            .as_ref()
            .and_then(|result| result.get("contract_version"))
            .and_then(|value| value.as_str()),
        Some(MOBKIT_CONTRACT_VERSION)
    );
    let methods = rpc_response
        .result
        .as_ref()
        .and_then(|result| result.get("methods"))
        .and_then(|value| value.as_array())
        .expect("capabilities methods");
    assert!(
        methods
            .iter()
            .any(|method| method == "mobkit/events/subscribe")
    );

    let python_request = run_shell(
        r#"
python3 - <<'PY'
import json
print(json.dumps({"jsonrpc":"2.0","id":"sdk-py-1","method":"mobkit/status","params":{}}))
PY
"#,
        &[],
    );
    let py_response_json =
        handle_mobkit_rpc_json(&mut runtime, python_request.trim(), Duration::from_secs(1));
    let py_validation = run_shell(
        r#"
python3 - <<'PY'
import json, os
resp = json.loads(os.environ["RPC_RESPONSE"])
assert resp["jsonrpc"] == "2.0"
assert resp["id"] == "sdk-py-1"
assert resp["result"]["contract_version"] == os.environ["EXPECTED_CONTRACT"]
print("PY_RESPONSE_VALIDATED:1")
PY
"#,
        &[
            ("RPC_RESPONSE", py_response_json.as_str()),
            ("EXPECTED_CONTRACT", MOBKIT_CONTRACT_VERSION),
        ],
    );
    assert!(py_validation.contains("PY_RESPONSE_VALIDATED:1"));

    let console_state = build_runtime_decision_state(RuntimeDecisionInputs {
        bigquery: BigQueryNaming {
            dataset: "phase0b_dataset".to_string(),
            table: "sessions".to_string(),
        },
        trusted_mobkit_toml: trusted_toml_for_sdk_console_probe(),
        auth: AuthPolicy {
            default_provider: AuthProvider::GoogleOAuth,
            email_allowlist: vec!["alice@example.com".to_string()],
        },
        trusted_oidc: trusted_oidc(),
        console: ConsolePolicy {
            require_app_auth: true,
        },
        ops: RuntimeOpsPolicy::default(),
        release_metadata_json: release_json(),
    })
    .expect("console decision state should build");

    let authorized = handle_console_rest_json_route(
        &console_state,
        &ConsoleRestJsonRequest {
            method: "GET".to_string(),
            path: "/console/modules".to_string(),
            auth: Some(ConsoleAccessRequest {
                provider: AuthProvider::GoogleOAuth,
                email: "alice@example.com".to_string(),
            }),
        },
    );
    assert_eq!(authorized.status, 200);
    assert_eq!(
        authorized.body["modules"].as_array().map(|v| v.len()),
        Some(2)
    );

    let unauthorized = handle_console_rest_json_route(
        &console_state,
        &ConsoleRestJsonRequest {
            method: "GET".to_string(),
            path: "/console/modules".to_string(),
            auth: None,
        },
    );
    assert_eq!(unauthorized.status, 401);

    let _ = runtime.shutdown();
}
