observer_core/

runtime.rs

// SPDX-FileCopyrightText: 2026 Alexander R. Croft
// SPDX-License-Identifier: GPL-3.0-or-later

use crate::error::{ObserverError, ObserverResult};
use crate::inventory::{Inventory, InventoryEntry, InventoryRunner};
use crate::normalize::{normalized_inventory_sha256, normalized_suite_sha256};
use crate::providers::{decode_b64_bytes, run_provider_target, ResolvedProviders};
use crate::report::{
    derive_case_id, preview_bytes, ActionFail, ActionRecord, ActionStatus, ArtifactRecord,
    AssertRecord, CaseRecord, ExtractRecord, ReportHeader, ReportMode, ReportRecord, Status, SummaryRecord,
    TelemetryEntry, TelemetryRecord,
};
use crate::suite::{
    CaseKeyField, CaseSource, CompareOp, Predicate, ResultExpr, SelectionMode, Selector, Statement,
    SuiteCore, ValueExpr,
};
use regex::Regex;
use serde_json::json;
use serde_json::Number;
use std::collections::BTreeMap;
use std::fs;
use std::io::{Read, Write};
use std::net::{SocketAddr, TcpStream, ToSocketAddrs};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::thread;
use std::time::{Duration, Instant};

#[derive(Debug, Clone)]
pub struct ExecutionOutcome {
    pub records: Vec<ReportRecord>,
    pub exit_code: i32,
}

#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ExecutionOptions {
    pub emit_telemetry: bool,
    pub filter_test_name: Option<String>,
}

#[derive(Debug, Clone, PartialEq)]
struct ProcRun {
    exit: i32,
    out: Vec<u8>,
    err: Vec<u8>,
    telemetry: Vec<TelemetryEntry>,
}

#[derive(Debug, Clone, PartialEq)]
struct HttpResp {
    status: i64,
    body: Vec<u8>,
    headers: BTreeMap<String, String>,
}

#[derive(Debug, Clone, PartialEq)]
struct TcpResp {
    bytes: Vec<u8>,
}

#[derive(Debug, Clone, PartialEq)]
struct RuntimeFail {
    kind: String,
    msg: String,
    code: Option<String>,
}

#[derive(Debug, Clone, PartialEq)]
enum RuntimeValue {
    Int(i64),
    String(String),
    Bytes(Vec<u8>),
    Object(BTreeMap<String, RuntimeValue>),
    ProcRun(ProcRun),
    HttpResp(HttpResp),
    TcpResp(TcpResp),
    Fail(RuntimeFail),
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct ArtifactBinding {
    name: String,
    kind: String,
    path: String,
}

#[derive(Debug, Clone, PartialEq)]
struct CaseInstance {
    case_key: String,
    case_value: RuntimeValue,
    test_name: String,
}

#[derive(Debug, Default)]
struct CaseState {
    bindings: BTreeMap<String, RuntimeValue>,
    artifacts: BTreeMap<String, ArtifactBinding>,
    action_ix: usize,
    artifact_ix: usize,
    extract_ix: usize,
    assert_ix: usize,
    assert_pass: usize,
    assert_fail: usize,
    unhandled_action_fail: usize,
    records: Vec<ReportRecord>,
}

pub fn execute_suite(
    inventory: &Inventory,
    suite: &SuiteCore,
    mode: ReportMode,
) -> ObserverResult<ExecutionOutcome> {
    execute_suite_with_options(inventory, suite, mode, ExecutionOptions::default())
}

pub fn execute_suite_with_options(
    inventory: &Inventory,
    suite: &SuiteCore,
    mode: ReportMode,
    options: ExecutionOptions,
) -> ObserverResult<ExecutionOutcome> {
    execute_suite_with_providers_and_options(inventory, suite, mode, None, options)
}

pub fn execute_suite_with_providers(
    inventory: &Inventory,
    suite: &SuiteCore,
    mode: ReportMode,
    providers: Option<&ResolvedProviders>,
) -> ObserverResult<ExecutionOutcome> {
    execute_suite_with_providers_and_options(
        inventory,
        suite,
        mode,
        providers,
        ExecutionOptions::default(),
    )
}

pub fn execute_suite_with_providers_and_options(
    inventory: &Inventory,
    suite: &SuiteCore,
    mode: ReportMode,
    providers: Option<&ResolvedProviders>,
    options: ExecutionOptions,
) -> ObserverResult<ExecutionOutcome> {
    let inventory_sha256 = normalized_inventory_sha256(inventory)?;
    let suite_sha256 = normalized_suite_sha256(suite)?;

    let mut records = vec![ReportRecord::Header(ReportHeader::new(
        inventory_sha256,
        suite_sha256,
        mode,
    ))];

    let mut case_pass = 0usize;
    let mut case_fail = 0usize;
    let mut assert_pass = 0usize;
    let mut assert_fail = 0usize;

    for item in &suite.items {
        let mut selected = discover_cases(inventory, &item.case_source)?;
        if let Some(filter) = options.filter_test_name.as_deref() {
            selected.retain(|case| case.case_key == filter || case.test_name == filter);
        }
        if selected.is_empty() && matches!(item.selection_mode, SelectionMode::Required) {
            if options.filter_test_name.is_some() {
                continue;
            }
            return Err(ObserverError::Runtime(format!(
                "required case source for item `{}` produced no cases",
                item.item_id
            )));
        }

        for case in selected {
            let case_id = derive_case_id(&item.item_id, &case.case_key);
            let mut state = CaseState::default();
            state
                .bindings
                .insert(item.case_binding.clone(), case.case_value.clone());

            for statement in &item.body {
                evaluate_statement(statement, inventory, providers, &case_id, &mut state, &options)?;
            }

            let status = if state.assert_fail == 0 && state.unhandled_action_fail == 0 {
                case_pass += 1;
                Status::Pass
            } else {
                case_fail += 1;
                Status::Fail
            };

            assert_pass += state.assert_pass;
            assert_fail += state.assert_fail;
            records.append(&mut state.records);
            records.push(ReportRecord::Case(CaseRecord {
                case_id,
                item_id: item.item_id.clone(),
                test_name: case.test_name,
                case_key: match item.case_source {
                    CaseSource::Inventory { .. } => None,
                    CaseSource::Files { .. } => Some(case.case_key),
                },
                status,
                assert_pass: state.assert_pass,
                assert_fail: state.assert_fail,
                unhandled_action_fail: state.unhandled_action_fail,
            }));
        }
    }

    let exit_code = if case_fail == 0 { 0 } else { 1 };
    records.push(ReportRecord::Summary(SummaryRecord {
        case_pass,
        case_fail,
        assert_pass,
        assert_fail,
        exit_code,
    }));

    Ok(ExecutionOutcome { records, exit_code })
}

fn discover_cases(inventory: &Inventory, case_source: &CaseSource) -> ObserverResult<Vec<CaseInstance>> {
    match case_source {
        CaseSource::Inventory { selector } => {
            let mut names = Vec::new();
            for entry in &inventory.entries {
                if matches_selector(&entry.name, selector)? {
                    names.push(entry.name.clone());
                }
            }
            names.sort_by(|left, right| left.as_bytes().cmp(right.as_bytes()));
            Ok(names
                .into_iter()
                .map(|name| CaseInstance {
                    case_key: name.clone(),
                    case_value: RuntimeValue::String(name.clone()),
                    test_name: name,
                })
                .collect())
        }
        CaseSource::Files {
            root,
            glob,
            key_field,
        } => discover_file_cases(root, glob, *key_field),
    }
}

fn matches_selector(name: &str, selector: &Selector) -> ObserverResult<bool> {
    match selector {
        Selector::Exact { value } => Ok(name == value),
        Selector::Prefix { value } => Ok(name.starts_with(value)),
        Selector::Glob { value } => Ok(matches_glob(name, value)),
        Selector::Regex { value } => Regex::new(value)
            .map(|regex| regex.is_match(name))
            .map_err(|error| ObserverError::Runtime(format!("invalid selector regex: {error}"))),
        Selector::Any { items } => {
            for item in items {
                if matches_selector(name, item)? {
                    return Ok(true);
                }
            }
            Ok(false)
        }
        Selector::All { items } => {
            for item in items {
                if !matches_selector(name, item)? {
                    return Ok(false);
                }
            }
            Ok(true)
        }
    }
}

fn matches_glob(name: &str, pattern: &str) -> bool {
    matches_glob_bytes(name.as_bytes(), pattern.as_bytes())
}

fn discover_file_cases(root: &str, glob: &str, key_field: CaseKeyField) -> ObserverResult<Vec<CaseInstance>> {
    let root_path = PathBuf::from(root);
    let mut rel_paths = Vec::new();
    collect_relative_files(&root_path, &root_path, &mut rel_paths)?;
    rel_paths.sort_by(|left, right| left.as_bytes().cmp(right.as_bytes()));

    let mut cases = Vec::new();
    for rel_path in rel_paths {
        if !matches_glob(&rel_path, glob) {
            continue;
        }
        let path = Path::new(&rel_path);
        let name = path
            .file_name()
            .and_then(|value| value.to_str())
            .ok_or_else(|| ObserverError::Runtime(format!("invalid discovered file name `{rel_path}`")))?
            .to_owned();
        let stem = path
            .file_stem()
            .and_then(|value| value.to_str())
            .ok_or_else(|| ObserverError::Runtime(format!("invalid discovered file stem `{rel_path}`")))?
            .to_owned();
        let ext = path
            .extension()
            .and_then(|value| value.to_str())
            .unwrap_or("")
            .to_owned();
        let case_key = match key_field {
            CaseKeyField::Path => rel_path.clone(),
            CaseKeyField::Name => name.clone(),
            CaseKeyField::Stem => stem.clone(),
        };
        let mut fields = BTreeMap::new();
        fields.insert("key".to_owned(), RuntimeValue::String(case_key.clone()));
        fields.insert("path".to_owned(), RuntimeValue::String(rel_path.clone()));
        fields.insert("name".to_owned(), RuntimeValue::String(name));
        fields.insert("stem".to_owned(), RuntimeValue::String(stem));
        fields.insert("ext".to_owned(), RuntimeValue::String(ext));
        cases.push(CaseInstance {
            case_key: case_key.clone(),
            test_name: case_key,
            case_value: RuntimeValue::Object(fields),
        });
    }
    Ok(cases)
}

fn collect_relative_files(root: &Path, current: &Path, out: &mut Vec<String>) -> ObserverResult<()> {
    let entries = fs::read_dir(current)
        .map_err(|error| ObserverError::Runtime(format!("failed to read `{}`: {error}", current.display())))?;
    let mut children = Vec::new();
    for entry in entries {
        let entry = entry
            .map_err(|error| ObserverError::Runtime(format!("failed to enumerate `{}`: {error}", current.display())))?;
        children.push(entry.path());
    }
    children.sort_by(|left, right| left.as_os_str().as_encoded_bytes().cmp(right.as_os_str().as_encoded_bytes()));

    for path in children {
        let metadata = fs::metadata(&path)
            .map_err(|error| ObserverError::Runtime(format!("failed to stat `{}`: {error}", path.display())))?;
        if metadata.is_dir() {
            collect_relative_files(root, &path, out)?;
        } else if metadata.is_file() {
            let rel = path
                .strip_prefix(root)
                .map_err(|error| ObserverError::Runtime(format!("failed to relativize `{}`: {error}", path.display())))?;
            out.push(normalize_relative_path(rel));
        }
    }
    Ok(())
}

fn normalize_relative_path(path: &Path) -> String {
    path.components()
        .map(|component| component.as_os_str().to_string_lossy().into_owned())
        .collect::<Vec<_>>()
        .join("/")
}

fn matches_glob_bytes(text: &[u8], pattern: &[u8]) -> bool {
    if pattern.is_empty() {
        return text.is_empty();
    }
    if pattern.len() >= 3 && pattern[0] == b'*' && pattern[1] == b'*' && pattern[2] == b'/' {
        if matches_glob_bytes(text, &pattern[3..]) {
            return true;
        }
    }
    match pattern[0] {
        b'*' => {
            for offset in 0..=text.len() {
                if matches_glob_bytes(&text[offset..], &pattern[1..]) {
                    return true;
                }
            }
            false
        }
        b'?' => !text.is_empty() && matches_glob_bytes(&text[1..], &pattern[1..]),
        byte => !text.is_empty() && text[0] == byte && matches_glob_bytes(&text[1..], &pattern[1..]),
    }
}

fn evaluate_statement(
    statement: &Statement,
    inventory: &Inventory,
    providers: Option<&ResolvedProviders>,
    case_id: &str,
    state: &mut CaseState,
    options: &ExecutionOptions,
) -> ObserverResult<()> {
    match statement {
        Statement::Assert { predicate } => {
            let (status, msg) = evaluate_assert(predicate, state)?;
            let record = ReportRecord::Assert(AssertRecord {
                case_id: case_id.to_owned(),
                assert_ix: state.assert_ix,
                status,
                msg,
            });
            state.assert_ix += 1;
            match status {
                Status::Pass => state.assert_pass += 1,
                Status::Fail => state.assert_fail += 1,
            }
            state.records.push(record);
            Ok(())
        }
        Statement::Publish {
            name,
            artifact_kind,
            path,
        } => {
            let action_ix = state.action_ix;
            state.action_ix += 1;
            let path = as_string(eval_value(path, &state.bindings, &state.artifacts)?)?;
            match validate_artifact_path(&path, artifact_kind) {
                Ok(()) => {
                    let binding = ArtifactBinding {
                        name: name.clone(),
                        kind: artifact_kind.clone(),
                        path: path.clone(),
                    };
                    state.artifacts.insert(name.clone(), binding);
                    state.records.push(ReportRecord::Action(ActionRecord {
                        case_id: case_id.to_owned(),
                        action_ix,
                        action: "artifact_publish".to_owned(),
                        status: ActionStatus::Ok,
                        args: json!({"name": name, "kind": artifact_kind, "path": path}),
                        ok: Some(json!({"name": name, "kind": artifact_kind, "path": path})),
                        fail: None,
                    }));
                    state.records.push(ReportRecord::Artifact(ArtifactRecord {
                        case_id: case_id.to_owned(),
                        artifact_ix: state.artifact_ix,
                        action_ix,
                        name: name.clone(),
                        event: "publish".to_owned(),
                        kind: artifact_kind.clone(),
                        status: ActionStatus::Ok,
                        location: Some(json!({"path": path})),
                        fail: None,
                    }));
                    state.artifact_ix += 1;
                    Ok(())
                }
                Err(fail) => {
                    state.records.push(ReportRecord::Action(ActionRecord {
                        case_id: case_id.to_owned(),
                        action_ix,
                        action: "artifact_publish".to_owned(),
                        status: ActionStatus::Fail,
                        args: json!({"name": name, "kind": artifact_kind, "path": path}),
                        ok: None,
                        fail: Some(ActionFail {
                            kind: fail.kind.clone(),
                            msg: fail.msg.clone(),
                            code: fail.code.clone(),
                        }),
                    }));
                    state.records.push(ReportRecord::Artifact(ArtifactRecord {
                        case_id: case_id.to_owned(),
                        artifact_ix: state.artifact_ix,
                        action_ix,
                        name: name.clone(),
                        event: "publish".to_owned(),
                        kind: artifact_kind.clone(),
                        status: ActionStatus::Fail,
                        location: Some(json!({"path": path})),
                        fail: Some(ActionFail {
                            kind: fail.kind.clone(),
                            msg: fail.msg.clone(),
                            code: fail.code.clone(),
                        }),
                    }));
                    state.artifact_ix += 1;
                    Err(ObserverError::Runtime(format!("{}: {}", fail.kind, fail.msg)))
                }
            }
        }
        Statement::ResultBranch {
            result,
            ok_binding,
            ok,
            fail_binding,
            fail,
        } => {
            match execute_result(result, inventory, providers, case_id, state, options) {
                Ok(value) => {
                    let previous = state.bindings.insert(ok_binding.clone(), value);
                    for nested in ok {
                        evaluate_statement(nested, inventory, providers, case_id, state, options)?;
                    }
                    restore_binding(&mut state.bindings, ok_binding, previous);
                }
                Err(runtime_fail) => {
                    if fail.is_empty() {
                        state.unhandled_action_fail += 1;
                    } else if let Some(binding) = fail_binding {
                        let previous = state
                            .bindings
                            .insert(binding.clone(), RuntimeValue::Fail(runtime_fail));
                        for nested in fail {
                            evaluate_statement(nested, inventory, providers, case_id, state, options)?;
                        }
                        restore_binding(&mut state.bindings, binding, previous);
                    }
                }
            }
            Ok(())
        }
        Statement::BoolBranch {
            predicate,
            if_true,
            if_false,
        } => {
            if evaluate_predicate(predicate, state)? {
                for nested in if_true {
                    evaluate_statement(nested, inventory, providers, case_id, state, options)?;
                }
            } else {
                for nested in if_false {
                    evaluate_statement(nested, inventory, providers, case_id, state, options)?;
                }
            }
            Ok(())
        }
    }
}

fn restore_binding(
    bindings: &mut BTreeMap<String, RuntimeValue>,
    key: &str,
    previous: Option<RuntimeValue>,
) {
    if let Some(value) = previous {
        bindings.insert(key.to_owned(), value);
    } else {
        bindings.remove(key);
    }
}

fn execute_result(
    result: &ResultExpr,
    inventory: &Inventory,
    providers: Option<&ResolvedProviders>,
    case_id: &str,
    state: &mut CaseState,
    options: &ExecutionOptions,
) -> Result<RuntimeValue, RuntimeFail> {
    match result {
        ResultExpr::Run { test, timeout_ms } => {
            let test_name = as_string(eval_value(test, &state.bindings, &state.artifacts).map_err(to_runtime_fail)?)
                .map_err(to_runtime_fail)?;
            let entry = inventory
                .entries
                .iter()
                .find(|entry| entry.name == test_name)
                .ok_or_else(|| RuntimeFail {
                    kind: "unknown-test".to_owned(),
                    msg: "unknown test".to_owned(),
                    code: None,
                })?;
            let action_ix = state.action_ix;
            state.action_ix += 1;
            match execute_inventory_entry(entry, providers, *timeout_ms) {
                Ok(run) => {
                    let telemetry = run.telemetry.clone();
                    state.records.push(ReportRecord::Action(ActionRecord {
                        case_id: case_id.to_owned(),
                        action_ix,
                        action: "run".to_owned(),
                        status: ActionStatus::Ok,
                        args: json!({"test_name": test_name, "timeout_ms": timeout_ms}),
                        ok: Some(proc_ok_payload(&run)),
                        fail: None,
                    }));
                    if options.emit_telemetry {
                        append_action_telemetry_records(&mut state.records, case_id, action_ix, &telemetry);
                    }
                    Ok(RuntimeValue::ProcRun(run))
                }
                Err(fail) => {
                    state.records.push(ReportRecord::Action(ActionRecord {
                        case_id: case_id.to_owned(),
                        action_ix,
                        action: "run".to_owned(),
                        status: ActionStatus::Fail,
                        args: json!({"test_name": test_name, "timeout_ms": timeout_ms}),
                        ok: None,
                        fail: Some(ActionFail {
                            kind: fail.kind.clone(),
                            msg: fail.msg.clone(),
                            code: fail.code.clone(),
                        }),
                    }));
                    Err(fail)
                }
            }
        }
        ResultExpr::Proc {
            path,
            args,
            timeout_ms,
        } => {
            let path = as_string(eval_value(path, &state.bindings, &state.artifacts).map_err(to_runtime_fail)?)
                .map_err(to_runtime_fail)?;
            let resolved_args = evaluate_proc_args(args, &state.bindings, &state.artifacts)?;
            let action_ix = state.action_ix;
            state.action_ix += 1;
            match execute_process(&path, &resolved_args, *timeout_ms) {
                Ok(run) => {
                    let telemetry = run.telemetry.clone();
                    state.records.push(ReportRecord::Action(ActionRecord {
                        case_id: case_id.to_owned(),
                        action_ix,
                        action: "proc".to_owned(),
                        status: ActionStatus::Ok,
                        args: json!({"path": path, "args": resolved_args, "timeout_ms": timeout_ms}),
                        ok: Some(proc_ok_payload(&run)),
                        fail: None,
                    }));
                    if options.emit_telemetry {
                        append_action_telemetry_records(&mut state.records, case_id, action_ix, &telemetry);
                    }
                    Ok(RuntimeValue::ProcRun(run))
                }
                Err(fail) => {
                    state.records.push(ReportRecord::Action(ActionRecord {
                        case_id: case_id.to_owned(),
                        action_ix,
                        action: "proc".to_owned(),
                        status: ActionStatus::Fail,
                        args: json!({"path": path, "args": resolved_args, "timeout_ms": timeout_ms}),
                        ok: None,
                        fail: Some(ActionFail {
                            kind: fail.kind.clone(),
                            msg: fail.msg.clone(),
                            code: fail.code.clone(),
                        }),
                    }));
                    Err(fail)
                }
            }
        }
        ResultExpr::HttpGet { url, timeout_ms } => {
            let url = as_string(eval_value(url, &state.bindings, &state.artifacts).map_err(to_runtime_fail)?)
                .map_err(to_runtime_fail)?;
            let action_ix = state.action_ix;
            state.action_ix += 1;
            match execute_http_get(&url, *timeout_ms) {
                Ok(response) => {
                    state.records.push(ReportRecord::Action(ActionRecord {
                        case_id: case_id.to_owned(),
                        action_ix,
                        action: "http_get".to_owned(),
                        status: ActionStatus::Ok,
                        args: json!({"url": url, "timeout_ms": timeout_ms}),
                        ok: Some(http_ok_payload(&response)),
                        fail: None,
                    }));
                    Ok(RuntimeValue::HttpResp(response))
                }
                Err(fail) => {
                    state.records.push(ReportRecord::Action(ActionRecord {
                        case_id: case_id.to_owned(),
                        action_ix,
                        action: "http_get".to_owned(),
                        status: ActionStatus::Fail,
                        args: json!({"url": url, "timeout_ms": timeout_ms}),
                        ok: None,
                        fail: Some(ActionFail {
                            kind: fail.kind.clone(),
                            msg: fail.msg.clone(),
                            code: fail.code.clone(),
                        }),
                    }));
                    Err(fail)
                }
            }
        }
        ResultExpr::Tcp {
            address,
            send,
            recv_max,
            timeout_ms,
        } => {
            let address = as_string(eval_value(address, &state.bindings, &state.artifacts).map_err(to_runtime_fail)?)
                .map_err(to_runtime_fail)?;
            let send = as_bytes(eval_value(send, &state.bindings, &state.artifacts).map_err(to_runtime_fail)?)
                .map_err(to_runtime_fail)?;
            let action_ix = state.action_ix;
            state.action_ix += 1;
            match execute_tcp(&address, &send, *recv_max as usize, *timeout_ms) {
                Ok(response) => {
                    state.records.push(ReportRecord::Action(ActionRecord {
                        case_id: case_id.to_owned(),
                        action_ix,
                        action: "tcp".to_owned(),
                        status: ActionStatus::Ok,
                        args: json!({
                            "address": address,
                            "send_len": send.len(),
                            "recv_max": recv_max,
                            "timeout_ms": timeout_ms
                        }),
                        ok: Some(tcp_ok_payload(&response)),
                        fail: None,
                    }));
                    Ok(RuntimeValue::TcpResp(response))
                }
                Err(fail) => {
                    state.records.push(ReportRecord::Action(ActionRecord {
                        case_id: case_id.to_owned(),
                        action_ix,
                        action: "tcp".to_owned(),
                        status: ActionStatus::Fail,
                        args: json!({
                            "address": address,
                            "send_len": send.len(),
                            "recv_max": recv_max,
                            "timeout_ms": timeout_ms
                        }),
                        ok: None,
                        fail: Some(ActionFail {
                            kind: fail.kind.clone(),
                            msg: fail.msg.clone(),
                            code: fail.code.clone(),
                        }),
                    }));
                    Err(fail)
                }
            }
        }
        ResultExpr::ArtifactCheck { name, artifact_kind } => {
            let action_ix = state.action_ix;
            state.action_ix += 1;
            match check_artifact(name, artifact_kind, &state.artifacts) {
                Ok(binding) => {
                    state.records.push(ReportRecord::Action(ActionRecord {
                        case_id: case_id.to_owned(),
                        action_ix,
                        action: "artifact_check".to_owned(),
                        status: ActionStatus::Ok,
                        args: json!({"name": name, "kind": artifact_kind}),
                        ok: Some(json!({"name": binding.name, "kind": binding.kind, "path": binding.path})),
                        fail: None,
                    }));
                    state.records.push(ReportRecord::Artifact(ArtifactRecord {
                        case_id: case_id.to_owned(),
                        artifact_ix: state.artifact_ix,
                        action_ix,
                        name: name.clone(),
                        event: "check".to_owned(),
                        kind: artifact_kind.clone(),
                        status: ActionStatus::Ok,
                        location: Some(json!({"path": binding.path})),
                        fail: None,
                    }));
                    state.artifact_ix += 1;
                    Ok(artifact_binding_value(&binding))
                }
                Err(fail) => {
                    state.records.push(ReportRecord::Action(ActionRecord {
                        case_id: case_id.to_owned(),
                        action_ix,
                        action: "artifact_check".to_owned(),
                        status: ActionStatus::Fail,
                        args: json!({"name": name, "kind": artifact_kind}),
                        ok: None,
                        fail: Some(ActionFail {
                            kind: fail.kind.clone(),
                            msg: fail.msg.clone(),
                            code: fail.code.clone(),
                        }),
                    }));
                    state.records.push(ReportRecord::Artifact(ArtifactRecord {
                        case_id: case_id.to_owned(),
                        artifact_ix: state.artifact_ix,
                        action_ix,
                        name: name.clone(),
                        event: "check".to_owned(),
                        kind: artifact_kind.clone(),
                        status: ActionStatus::Fail,
                        location: None,
                        fail: Some(ActionFail {
                            kind: fail.kind.clone(),
                            msg: fail.msg.clone(),
                            code: fail.code.clone(),
                        }),
                    }));
                    state.artifact_ix += 1;
                    Err(fail)
                }
            }
        }
        ResultExpr::ExtractJson { name, select } => {
            execute_extract(name, select, false, case_id, state)
        }
        ResultExpr::ExtractJsonl { name, select } => {
            execute_extract(name, select, true, case_id, state)
        }
    }
}

fn evaluate_proc_args(
    args: &[ValueExpr],
    bindings: &BTreeMap<String, RuntimeValue>,
    artifacts: &BTreeMap<String, ArtifactBinding>,
) -> Result<Vec<String>, RuntimeFail> {
    let mut resolved = Vec::with_capacity(args.len());
    for arg in args {
        let value = eval_value(arg, bindings, artifacts).map_err(to_runtime_fail)?;
        resolved.push(as_string(value).map_err(to_runtime_fail)?);
    }
    Ok(resolved)
}

fn validate_artifact_path(path: &str, artifact_kind: &str) -> Result<(), RuntimeFail> {
    let metadata = fs::metadata(path).map_err(|error| RuntimeFail {
        kind: "artifact-missing".to_owned(),
        msg: format!("artifact `{path}` is missing: {error}"),
        code: None,
    })?;
    match artifact_kind {
        "directory" => {
            if metadata.is_dir() {
                Ok(())
            } else {
                Err(RuntimeFail {
                    kind: "artifact-kind".to_owned(),
                    msg: format!("artifact `{path}` is not a directory"),
                    code: None,
                })
            }
        }
        _ => {
            if metadata.is_file() {
                Ok(())
            } else {
                Err(RuntimeFail {
                    kind: "artifact-kind".to_owned(),
                    msg: format!("artifact `{path}` is not a file"),
                    code: None,
                })
            }
        }
    }
}

fn check_artifact(
    name: &str,
    artifact_kind: &str,
    artifacts: &BTreeMap<String, ArtifactBinding>,
) -> Result<ArtifactBinding, RuntimeFail> {
    let binding = artifacts.get(name).cloned().ok_or_else(|| RuntimeFail {
        kind: "artifact-missing".to_owned(),
        msg: format!("unknown artifact `{name}`"),
        code: None,
    })?;
    if binding.kind != artifact_kind {
        return Err(RuntimeFail {
            kind: "artifact-kind".to_owned(),
            msg: format!(
                "artifact `{name}` has kind `{}`, expected `{artifact_kind}`",
                binding.kind
            ),
            code: None,
        });
    }
    validate_artifact_path(&binding.path, artifact_kind)?;
    Ok(binding)
}

fn artifact_binding_value(binding: &ArtifactBinding) -> RuntimeValue {
    RuntimeValue::Object(BTreeMap::from([
        ("name".to_owned(), RuntimeValue::String(binding.name.clone())),
        ("kind".to_owned(), RuntimeValue::String(binding.kind.clone())),
        ("path".to_owned(), RuntimeValue::String(binding.path.clone())),
    ]))
}

fn execute_extract(
    name: &str,
    select: &str,
    jsonl: bool,
    case_id: &str,
    state: &mut CaseState,
) -> Result<RuntimeValue, RuntimeFail> {
    let binding = check_artifact(name, if jsonl { "jsonl" } else { "json" }, &state.artifacts)?;
    let action_ix = state.action_ix;
    state.action_ix += 1;
    match extract_value(&binding.path, select, jsonl) {
        Ok((match_count, value_text)) => {
            let action_name = if jsonl { "extract_jsonl" } else { "extract_json" };
            let format = if jsonl { "jsonl" } else { "json" };
            let (preview_len, preview_b64, truncated) = preview_bytes(value_text.as_bytes());
            let value_text_field = if preview_len == 0 { None } else { Some(value_text.clone()) };
            state.records.push(ReportRecord::Action(ActionRecord {
                case_id: case_id.to_owned(),
                action_ix,
                action: action_name.to_owned(),
                status: ActionStatus::Ok,
                args: json!({"name": name, "select": select}),
                ok: Some(json!({
                    "source_artifact": name,
                    "select": select,
                    "match_count": match_count,
                    "value_text": value_text_field,
                    "value_text_truncated": truncated
                })),
                fail: None,
            }));
            state.records.push(ReportRecord::Extract(ExtractRecord {
                case_id: case_id.to_owned(),
                extract_ix: state.extract_ix,
                action_ix,
                source_artifact: name.to_owned(),
                format: format.to_owned(),
                status: ActionStatus::Ok,
                select: Some(select.to_owned()),
                summary: Some(json!({
                    "match_count": match_count,
                    "value_text": preview_b64.map(|_| value_text),
                    "value_text_truncated": truncated
                })),
                fail: None,
            }));
            state.extract_ix += 1;
            Ok(RuntimeValue::Object(BTreeMap::from([
                ("format".to_owned(), RuntimeValue::String(format.to_owned())),
                ("select".to_owned(), RuntimeValue::String(select.to_owned())),
                ("count".to_owned(), RuntimeValue::Int(match_count as i64)),
                ("text".to_owned(), RuntimeValue::String(value_text_field.unwrap_or_default())),
            ])))
        }
        Err(fail) => {
            let action_name = if jsonl { "extract_jsonl" } else { "extract_json" };
            let format = if jsonl { "jsonl" } else { "json" };
            state.records.push(ReportRecord::Action(ActionRecord {
                case_id: case_id.to_owned(),
                action_ix,
                action: action_name.to_owned(),
                status: ActionStatus::Fail,
                args: json!({"name": name, "select": select}),
                ok: None,
                fail: Some(ActionFail {
                    kind: fail.kind.clone(),
                    msg: fail.msg.clone(),
                    code: fail.code.clone(),
                }),
            }));
            state.records.push(ReportRecord::Extract(ExtractRecord {
                case_id: case_id.to_owned(),
                extract_ix: state.extract_ix,
                action_ix,
                source_artifact: name.to_owned(),
                format: format.to_owned(),
                status: ActionStatus::Fail,
                select: Some(select.to_owned()),
                summary: None,
                fail: Some(ActionFail {
                    kind: fail.kind.clone(),
                    msg: fail.msg.clone(),
                    code: fail.code.clone(),
                }),
            }));
            state.extract_ix += 1;
            Err(fail)
        }
    }
}

fn extract_value(path: &str, select: &str, jsonl: bool) -> Result<(usize, String), RuntimeFail> {
    let source = fs::read_to_string(path).map_err(|error| RuntimeFail {
        kind: "io".to_owned(),
        msg: format!("failed to read `{path}`: {error}"),
        code: None,
    })?;
    let mut matches = Vec::new();
    if jsonl {
        for line in source.lines() {
            if line.trim().is_empty() {
                continue;
            }
            let value = serde_json::from_str::<serde_json::Value>(line).map_err(|error| RuntimeFail {
                kind: "protocol".to_owned(),
                msg: format!("invalid jsonl in `{path}`: {error}"),
                code: None,
            })?;
            if let Some(found) = select_json_value(&value, select) {
                matches.push(found);
            }
        }
    } else {
        let value = serde_json::from_str::<serde_json::Value>(&source).map_err(|error| RuntimeFail {
            kind: "protocol".to_owned(),
            msg: format!("invalid json in `{path}`: {error}"),
            code: None,
        })?;
        if let Some(found) = select_json_value(&value, select) {
            matches.push(found);
        }
    }
    let first = matches.first().cloned().unwrap_or_else(|| "".to_owned());
    Ok((matches.len(), first))
}

fn select_json_value(value: &serde_json::Value, select: &str) -> Option<String> {
    let path = select.strip_prefix("$.")?;
    let mut current = value;
    for segment in path.split('.') {
        current = current.get(segment)?;
    }
    Some(match current {
        serde_json::Value::String(text) => text.clone(),
        other => other.to_string(),
    })
}

fn normalize_joined_path(parts: &[String]) -> String {
    let mut path = PathBuf::new();
    for part in parts {
        path.push(part);
    }
    normalize_relative_path(&path)
}

fn execute_inventory_entry(
    entry: &InventoryEntry,
    providers: Option<&ResolvedProviders>,
    timeout_ms: u32,
) -> Result<ProcRun, RuntimeFail> {
    match &entry.runner {
        InventoryRunner::Exec { path, args } => execute_process(path, args, timeout_ms),
        InventoryRunner::Sh { command } => execute_process("/bin/sh", &["-c".to_owned(), command.clone()], timeout_ms),
        InventoryRunner::Provider { provider, target } => {
            let providers = providers.ok_or_else(|| RuntimeFail {
                kind: "provider-config".to_owned(),
                msg: "provider config required".to_owned(),
                code: None,
            })?;
            let resolved = providers.get(provider).ok_or_else(|| RuntimeFail {
                kind: "provider-config".to_owned(),
                msg: format!("unresolved provider `{provider}`"),
                code: None,
            })?;
            let response = run_provider_target(resolved, target, timeout_ms).map_err(to_runtime_fail)?;
            let out = decode_b64_bytes(&response.out_b64).map_err(to_runtime_fail)?;
            let err = decode_b64_bytes(&response.err_b64).map_err(to_runtime_fail)?;
            Ok(ProcRun {
                exit: response.exit,
                out,
                err,
                telemetry: response.telemetry,
            })
        }
    }
}

fn execute_process(path: &str, args: &[String], timeout_ms: u32) -> Result<ProcRun, RuntimeFail> {
    let started_at = Instant::now();
    let usage_before = child_resource_usage();
    let mut child = Command::new(path)
        .args(args)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .map_err(|error| RuntimeFail {
            kind: "spawn".to_owned(),
            msg: error.to_string(),
            code: None,
        })?;

    let deadline = Instant::now() + Duration::from_millis(u64::from(timeout_ms));
    loop {
        match child.try_wait() {
            Ok(Some(_)) => {
                let output = child.wait_with_output().map_err(|error| RuntimeFail {
                    kind: "io".to_owned(),
                    msg: error.to_string(),
                    code: None,
                })?;
                let usage_after = child_resource_usage();
                return Ok(ProcRun {
                    exit: output.status.code().unwrap_or(-1),
                    out: output.stdout,
                    err: output.stderr,
                    telemetry: runner_telemetry(started_at, usage_before, usage_after),
                });
            }
            Ok(None) => {
                if Instant::now() >= deadline {
                    let _ = child.kill();
                    let _ = child.wait();
                    return Err(RuntimeFail {
                        kind: "timeout".to_owned(),
                        msg: "timeout".to_owned(),
                        code: None,
                    });
                }
                thread::sleep(Duration::from_millis(10));
            }
            Err(error) => {
                return Err(RuntimeFail {
                    kind: "io".to_owned(),
                    msg: error.to_string(),
                    code: None,
                })
            }
        }
    }
}

fn runner_telemetry(
    started_at: Instant,
    usage_before: Option<ChildResourceUsage>,
    usage_after: Option<ChildResourceUsage>,
) -> Vec<TelemetryEntry> {
    let mut telemetry = vec![TelemetryEntry {
        name: "wall_time_ns".to_owned(),
        unit: Some("ns".to_owned()),
        value: crate::TelemetryValue::Metric {
            value: telemetry_number_from_u128(started_at.elapsed().as_nanos()),
        },
    }];

    if let (Some(before), Some(after)) = (usage_before, usage_after) {
        telemetry.push(TelemetryEntry {
            name: "cpu_user_ns".to_owned(),
            unit: Some("ns".to_owned()),
            value: crate::TelemetryValue::Metric {
                value: Number::from(after.user_ns.saturating_sub(before.user_ns)),
            },
        });
        telemetry.push(TelemetryEntry {
            name: "cpu_system_ns".to_owned(),
            unit: Some("ns".to_owned()),
            value: crate::TelemetryValue::Metric {
                value: Number::from(after.system_ns.saturating_sub(before.system_ns)),
            },
        });
        telemetry.push(TelemetryEntry {
            name: "peak_rss_bytes".to_owned(),
            unit: Some("bytes".to_owned()),
            value: crate::TelemetryValue::Metric {
                value: Number::from(after.peak_rss_bytes),
            },
        });
    }

    telemetry
}

fn telemetry_number_from_u128(value: u128) -> Number {
    let value = value.min(u128::from(u64::MAX));
    Number::from(value as u64)
}

#[derive(Debug, Clone, Copy)]
struct ChildResourceUsage {
    user_ns: u64,
    system_ns: u64,
    peak_rss_bytes: u64,
}

#[cfg(unix)]
fn child_resource_usage() -> Option<ChildResourceUsage> {
    let mut usage = std::mem::MaybeUninit::<libc::rusage>::zeroed();
    let status = unsafe { libc::getrusage(libc::RUSAGE_CHILDREN, usage.as_mut_ptr()) };
    if status != 0 {
        return None;
    }

    let usage = unsafe { usage.assume_init() };
    Some(ChildResourceUsage {
        user_ns: timeval_to_ns(usage.ru_utime),
        system_ns: timeval_to_ns(usage.ru_stime),
        peak_rss_bytes: peak_rss_bytes(usage.ru_maxrss),
    })
}

#[cfg(not(unix))]
fn child_resource_usage() -> Option<ChildResourceUsage> {
    None
}

#[cfg(unix)]
fn timeval_to_ns(value: libc::timeval) -> u64 {
    let secs = u64::try_from(value.tv_sec).unwrap_or(0);
    let micros = u64::try_from(value.tv_usec).unwrap_or(0);
    secs.saturating_mul(1_000_000_000)
        .saturating_add(micros.saturating_mul(1_000))
}

#[cfg(unix)]
fn peak_rss_bytes(value: libc::c_long) -> u64 {
    let value = u64::try_from(value).unwrap_or(0);
    #[cfg(target_os = "linux")]
    {
        value.saturating_mul(1024)
    }
    #[cfg(not(target_os = "linux"))]
    {
        value
    }
}

fn append_action_telemetry_records(
    records: &mut Vec<ReportRecord>,
    case_id: &str,
    action_ix: usize,
    telemetry: &[TelemetryEntry],
) {
    for entry in telemetry {
        records.push(ReportRecord::Telemetry(TelemetryRecord::action(
            case_id.to_owned(),
            action_ix,
            entry.clone(),
        )));
    }
}

fn evaluate_assert(predicate: &Predicate, state: &CaseState) -> ObserverResult<(Status, String)> {
    match predicate {
        Predicate::Fail { msg } => Ok((Status::Fail, msg.clone())),
        _ => {
            let pass = evaluate_predicate(predicate, state)?;
            if pass {
                Ok((Status::Pass, "expectation passed".to_owned()))
            } else {
                Ok((Status::Fail, "expectation failed".to_owned()))
            }
        }
    }
}

fn evaluate_predicate(predicate: &Predicate, state: &CaseState) -> ObserverResult<bool> {
    match predicate {
        Predicate::Compare { op, left, right } => {
            let left = eval_value(left, &state.bindings, &state.artifacts)?;
            let right = eval_value(right, &state.bindings, &state.artifacts)?;
            compare_values(*op, &left, &right)
        }
        Predicate::IsStatus { left, status } => {
            let left = eval_value(left, &state.bindings, &state.artifacts)?;
            is_status_value(&left, *status)
        }
        Predicate::IsStatusClass { left, class } => {
            let left = eval_value(left, &state.bindings, &state.artifacts)?;
            is_status_class_value(&left, *class)
        }
        Predicate::HasHeader { left, name } => {
            let left = eval_value(left, &state.bindings, &state.artifacts)?;
            has_header_value(&left, name)
        }
        Predicate::Contains { left, right } => {
            let left = eval_value(left, &state.bindings, &state.artifacts)?;
            let right = eval_value(right, &state.bindings, &state.artifacts)?;
            contains_value(&left, &right)
        }
        Predicate::ContainsRegex { left, regex } | Predicate::Match { left, regex } => {
            let left = eval_value(left, &state.bindings, &state.artifacts)?;
            match left {
                RuntimeValue::String(value) => Ok(Regex::new(regex)
                    .map_err(|error| ObserverError::Runtime(error.to_string()))?
                    .is_match(&value)),
                RuntimeValue::Bytes(bytes) => match String::from_utf8(bytes) {
                    Ok(value) => Ok(Regex::new(regex)
                        .map_err(|error| ObserverError::Runtime(error.to_string()))?
                        .is_match(&value)),
                    Err(_) => Ok(false),
                },
                _ => Err(ObserverError::Runtime(
                    "regex predicates require string or bytes operands".to_owned(),
                )),
            }
        }
        Predicate::StartsWith { left, right } => {
            let left = eval_value(left, &state.bindings, &state.artifacts)?;
            let right = eval_value(right, &state.bindings, &state.artifacts)?;
            starts_with_value(&left, &right)
        }
        Predicate::EndsWith { left, right } => {
            let left = eval_value(left, &state.bindings, &state.artifacts)?;
            let right = eval_value(right, &state.bindings, &state.artifacts)?;
            ends_with_value(&left, &right)
        }
        Predicate::Fail { .. } => Ok(false),
    }
}

fn compare_values(op: CompareOp, left: &RuntimeValue, right: &RuntimeValue) -> ObserverResult<bool> {
    match (left, right) {
        (RuntimeValue::Int(left), RuntimeValue::Int(right)) => Ok(match op {
            CompareOp::Eq => left == right,
            CompareOp::Ne => left != right,
            CompareOp::Lt => left < right,
            CompareOp::Le => left <= right,
            CompareOp::Gt => left > right,
            CompareOp::Ge => left >= right,
        }),
        (RuntimeValue::String(left), RuntimeValue::String(right)) => Ok(match op {
            CompareOp::Eq => left == right,
            CompareOp::Ne => left != right,
            _ => false,
        }),
        (RuntimeValue::Bytes(left), RuntimeValue::Bytes(right)) => Ok(match op {
            CompareOp::Eq => left == right,
            CompareOp::Ne => left != right,
            _ => false,
        }),
        (RuntimeValue::Bytes(left), RuntimeValue::String(right)) => compare_values(
            op,
            &RuntimeValue::Bytes(left.clone()),
            &RuntimeValue::Bytes(right.as_bytes().to_vec()),
        ),
        (RuntimeValue::String(left), RuntimeValue::Bytes(right)) => compare_values(
            op,
            &RuntimeValue::Bytes(left.as_bytes().to_vec()),
            &RuntimeValue::Bytes(right.clone()),
        ),
        _ => Err(ObserverError::Runtime(
            "unsupported operand types for comparison".to_owned(),
        )),
    }
}

fn contains_value(left: &RuntimeValue, right: &RuntimeValue) -> ObserverResult<bool> {
    match (left, right) {
        (RuntimeValue::String(left), RuntimeValue::String(right)) => Ok(left.contains(right)),
        (RuntimeValue::Bytes(left), RuntimeValue::Bytes(right)) => {
            Ok(left.windows(right.len()).any(|window| window == right.as_slice()))
        }
        (RuntimeValue::Bytes(left), RuntimeValue::String(right)) => contains_value(
            &RuntimeValue::Bytes(left.clone()),
            &RuntimeValue::Bytes(right.as_bytes().to_vec()),
        ),
        (RuntimeValue::String(left), RuntimeValue::Bytes(right)) => contains_value(
            &RuntimeValue::Bytes(left.as_bytes().to_vec()),
            &RuntimeValue::Bytes(right.clone()),
        ),
        _ => Err(ObserverError::Runtime(
            "unsupported operand types for contains".to_owned(),
        )),
    }
}

fn starts_with_value(left: &RuntimeValue, right: &RuntimeValue) -> ObserverResult<bool> {
    match (left, right) {
        (RuntimeValue::String(left), RuntimeValue::String(right)) => Ok(left.starts_with(right)),
        (RuntimeValue::Bytes(left), RuntimeValue::Bytes(right)) => Ok(left.starts_with(right)),
        (RuntimeValue::Bytes(left), RuntimeValue::String(right)) => starts_with_value(
            &RuntimeValue::Bytes(left.clone()),
            &RuntimeValue::Bytes(right.as_bytes().to_vec()),
        ),
        (RuntimeValue::String(left), RuntimeValue::Bytes(right)) => starts_with_value(
            &RuntimeValue::Bytes(left.as_bytes().to_vec()),
            &RuntimeValue::Bytes(right.clone()),
        ),
        _ => Err(ObserverError::Runtime(
            "unsupported operand types for startsWith".to_owned(),
        )),
    }
}

fn ends_with_value(left: &RuntimeValue, right: &RuntimeValue) -> ObserverResult<bool> {
    match (left, right) {
        (RuntimeValue::String(left), RuntimeValue::String(right)) => Ok(left.ends_with(right)),
        (RuntimeValue::Bytes(left), RuntimeValue::Bytes(right)) => Ok(left.ends_with(right)),
        (RuntimeValue::Bytes(left), RuntimeValue::String(right)) => ends_with_value(
            &RuntimeValue::Bytes(left.clone()),
            &RuntimeValue::Bytes(right.as_bytes().to_vec()),
        ),
        (RuntimeValue::String(left), RuntimeValue::Bytes(right)) => ends_with_value(
            &RuntimeValue::Bytes(left.as_bytes().to_vec()),
            &RuntimeValue::Bytes(right.clone()),
        ),
        _ => Err(ObserverError::Runtime(
            "unsupported operand types for endsWith".to_owned(),
        )),
    }
}

fn has_header_value(left: &RuntimeValue, name: &str) -> ObserverResult<bool> {
    let header_name = name.to_ascii_lowercase();
    match left {
        RuntimeValue::HttpResp(response) => Ok(response.headers.contains_key(&header_name)),
        _ => Err(ObserverError::Runtime(
            "hasHeader predicates require an http response operand".to_owned(),
        )),
    }
}

fn is_status_value(left: &RuntimeValue, status: i64) -> ObserverResult<bool> {
    match left {
        RuntimeValue::HttpResp(response) => Ok(response.status == status),
        _ => Err(ObserverError::Runtime(
            "isStatus predicates require an http response operand".to_owned(),
        )),
    }
}

fn is_status_class_value(left: &RuntimeValue, class: i64) -> ObserverResult<bool> {
    if !(1..=5).contains(&class) {
        return Err(ObserverError::Runtime(
            "isStatusClass requires a status class in the range 1..=5".to_owned(),
        ));
    }
    match left {
        RuntimeValue::HttpResp(response) => Ok(response.status / 100 == class),
        _ => Err(ObserverError::Runtime(
            "isStatusClass predicates require an http response operand".to_owned(),
        )),
    }
}

fn eval_value(
    expr: &ValueExpr,
    bindings: &BTreeMap<String, RuntimeValue>,
    artifacts: &BTreeMap<String, ArtifactBinding>,
) -> ObserverResult<RuntimeValue> {
    match expr {
        ValueExpr::Binding { name } => bindings
            .get(name)
            .cloned()
            .ok_or_else(|| ObserverError::Runtime(format!("unbound identifier `{name}`"))),
        ValueExpr::String { value } => Ok(RuntimeValue::String(value.clone())),
        ValueExpr::BytesUtf8 { value } => Ok(RuntimeValue::Bytes(value.as_bytes().to_vec())),
        ValueExpr::Int { value } => Ok(RuntimeValue::Int(*value)),
        ValueExpr::Field { base, name } => {
            let value = eval_value(base, bindings, artifacts)?;
            match (value, name.as_str()) {
                (RuntimeValue::ProcRun(run), "exit") => Ok(RuntimeValue::Int(run.exit.into())),
                (RuntimeValue::ProcRun(run), "out") => Ok(RuntimeValue::Bytes(run.out)),
                (RuntimeValue::ProcRun(run), "err") => Ok(RuntimeValue::Bytes(run.err)),
                (RuntimeValue::HttpResp(response), "status") => Ok(RuntimeValue::Int(response.status)),
                (RuntimeValue::HttpResp(response), "body") => Ok(RuntimeValue::Bytes(response.body)),
                (RuntimeValue::TcpResp(response), "bytes") => Ok(RuntimeValue::Bytes(response.bytes)),
                (RuntimeValue::Object(fields), field_name) => fields.get(field_name).cloned().ok_or_else(|| {
                    ObserverError::Runtime(format!("unknown field `{field_name}`"))
                }),
                (RuntimeValue::Fail(fail), "kind") => Ok(RuntimeValue::String(fail.kind)),
                (RuntimeValue::Fail(fail), "msg") => Ok(RuntimeValue::String(fail.msg)),
                (RuntimeValue::Fail(fail), "code") => {
                    if let Some(code) = fail.code {
                        Ok(RuntimeValue::String(code))
                    } else {
                        Err(ObserverError::Runtime(
                            "optional fail field `code` is absent".to_owned(),
                        ))
                    }
                }
                _ => Err(ObserverError::Runtime(format!("unknown field `{name}`"))),
            }
        }
        ValueExpr::Header { base, name } => {
            let header_name = name.to_ascii_lowercase();
            let value = eval_value(base, bindings, artifacts)?;
            match value {
                RuntimeValue::HttpResp(response) => response
                    .headers
                    .get(&header_name)
                    .cloned()
                    .map(|value| RuntimeValue::String(value))
                    .ok_or_else(|| {
                        ObserverError::Runtime(format!(
                            "header `{header_name}` is absent"
                        ))
                    }),
                _ => Err(ObserverError::Runtime(
                    "header lookup requires an http response operand".to_owned(),
                )),
            }
        }
        ValueExpr::ArtifactPath { name } => artifacts
            .get(name)
            .map(|binding| RuntimeValue::String(binding.path.clone()))
            .ok_or_else(|| ObserverError::Runtime(format!("unknown artifact `{name}`"))),
        ValueExpr::JoinPath { parts } => {
            let mut strings = Vec::with_capacity(parts.len());
            for part in parts {
                let value = eval_value(part, bindings, artifacts)?;
                strings.push(as_string(value)?);
            }
            Ok(RuntimeValue::String(normalize_joined_path(&strings)))
        }
    }
}

fn as_string(value: RuntimeValue) -> ObserverResult<String> {
    match value {
        RuntimeValue::String(value) => Ok(value),
        _ => Err(ObserverError::Runtime("expected string value".to_owned())),
    }
}

fn as_bytes(value: RuntimeValue) -> ObserverResult<Vec<u8>> {
    match value {
        RuntimeValue::Bytes(value) => Ok(value),
        RuntimeValue::String(value) => Ok(value.into_bytes()),
        _ => Err(ObserverError::Runtime("expected bytes or string value".to_owned())),
    }
}

fn proc_ok_payload(run: &ProcRun) -> serde_json::Value {
    let (out_len, out_preview, out_truncated) = preview_bytes(&run.out);
    let (err_len, err_preview, err_truncated) = preview_bytes(&run.err);
    let mut payload = json!({
        "exit": run.exit,
        "out_len": out_len,
        "err_len": err_len,
    });
    if let Some(preview) = out_preview {
        payload["out_preview_b64"] = json!(preview);
    }
    if let Some(preview) = err_preview {
        payload["err_preview_b64"] = json!(preview);
    }
    if let Some(truncated) = out_truncated {
        payload["out_truncated"] = json!(truncated);
    }
    if let Some(truncated) = err_truncated {
        payload["err_truncated"] = json!(truncated);
    }
    payload
}

fn http_ok_payload(response: &HttpResp) -> serde_json::Value {
    let (body_len, body_preview, body_truncated) = preview_bytes(&response.body);
    let mut payload = json!({
        "status": response.status,
        "body_len": body_len,
        "headers": response.headers,
    });
    if let Some(preview) = body_preview {
        payload["body_preview_b64"] = json!(preview);
    }
    if let Some(truncated) = body_truncated {
        payload["body_truncated"] = json!(truncated);
    }
    payload
}

fn tcp_ok_payload(response: &TcpResp) -> serde_json::Value {
    let (bytes_len, bytes_preview, bytes_truncated) = preview_bytes(&response.bytes);
    let mut payload = json!({
        "bytes_len": bytes_len,
    });
    if let Some(preview) = bytes_preview {
        payload["bytes_preview_b64"] = json!(preview);
    }
    if let Some(truncated) = bytes_truncated {
        payload["bytes_truncated"] = json!(truncated);
    }
    payload
}

fn execute_http_get(url: &str, timeout_ms: u32) -> Result<HttpResp, RuntimeFail> {
    let client = reqwest::blocking::Client::builder()
        .timeout(Duration::from_millis(u64::from(timeout_ms)))
        .build()
        .map_err(|error| RuntimeFail {
            kind: "http".to_owned(),
            msg: error.to_string(),
            code: None,
        })?;

    let response = client.get(url).send().map_err(http_error_to_runtime_fail)?;
    let status = i64::from(response.status().as_u16());
    let mut headers = BTreeMap::new();
    for (name, value) in response.headers() {
        let header_value = value.to_str().map_err(|error| RuntimeFail {
            kind: "http".to_owned(),
            msg: error.to_string(),
            code: None,
        })?;
        headers.insert(name.as_str().to_ascii_lowercase(), header_value.to_owned());
    }
    let body = response.bytes().map_err(http_error_to_runtime_fail)?.to_vec();

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

fn execute_tcp(
    address: &str,
    send: &[u8],
    recv_max: usize,
    timeout_ms: u32,
) -> Result<TcpResp, RuntimeFail> {
    let timeout = Duration::from_millis(u64::from(timeout_ms));
    let addrs = resolve_socket_addrs(address)?;
    let mut last_error = None;

    for socket_addr in addrs {
        match TcpStream::connect_timeout(&socket_addr, timeout) {
            Ok(mut stream) => {
                stream
                    .set_read_timeout(Some(timeout))
                    .map_err(io_runtime_fail)?;
                stream
                    .set_write_timeout(Some(timeout))
                    .map_err(io_runtime_fail)?;
                stream.write_all(send).map_err(io_runtime_fail)?;

                let mut bytes = Vec::new();
                let mut buffer = vec![0u8; recv_max.min(4096).max(1)];
                while bytes.len() < recv_max {
                    let remaining = recv_max - bytes.len();
                    let read_len = remaining.min(buffer.len());
                    match stream.read(&mut buffer[..read_len]) {
                        Ok(0) => break,
                        Ok(count) => bytes.extend_from_slice(&buffer[..count]),
                        Err(error)
                            if error.kind() == std::io::ErrorKind::WouldBlock
                                || error.kind() == std::io::ErrorKind::TimedOut =>
                        {
                            if bytes.is_empty() {
                                return Err(RuntimeFail {
                                    kind: "timeout".to_owned(),
                                    msg: "timeout".to_owned(),
                                    code: None,
                                });
                            }
                            break;
                        }
                        Err(error) => return Err(io_runtime_fail(error)),
                    }
                }

                return Ok(TcpResp { bytes });
            }
            Err(error) => last_error = Some(error),
        }
    }

    Err(last_error.map(io_runtime_fail).unwrap_or(RuntimeFail {
        kind: "tcp".to_owned(),
        msg: format!("no socket addresses resolved for `{address}`"),
        code: None,
    }))
}

fn resolve_socket_addrs(address: &str) -> Result<Vec<SocketAddr>, RuntimeFail> {
    let mut addrs = address.to_socket_addrs().map_err(io_runtime_fail)?.collect::<Vec<_>>();
    addrs.sort_by(|left, right| left.to_string().as_bytes().cmp(right.to_string().as_bytes()));
    addrs.dedup();
    Ok(addrs)
}

fn http_error_to_runtime_fail(error: reqwest::Error) -> RuntimeFail {
    let kind = if error.is_timeout() {
        "timeout"
    } else {
        "http"
    };
    RuntimeFail {
        kind: kind.to_owned(),
        msg: error.to_string(),
        code: error.status().map(|status| status.as_u16().to_string()),
    }
}

fn io_runtime_fail(error: std::io::Error) -> RuntimeFail {
    RuntimeFail {
        kind: "io".to_owned(),
        msg: error.to_string(),
        code: None,
    }
}

fn to_runtime_fail(error: ObserverError) -> RuntimeFail {
    match error {
        ObserverError::Runtime(msg) if msg == "timeout" => RuntimeFail {
            kind: "timeout".to_owned(),
            msg,
            code: None,
        },
        other => RuntimeFail {
            kind: "runtime".to_owned(),
            msg: other.to_string(),
            code: None,
        },
    }
}