harn-vm 0.10.132

//! Typed deserialization of host-emitted agent events.
//!
//! `__host_agent_emit_event` receives an untyped `(event_type, payload)`
//! pair from the Harn agent loop and turns it into a typed [`AgentEvent`].
//! Historically this lived in a ~570-line hand-written
//! `match event_type.as_str()` in `llm::agent_session_host` that
//! re-derived, field by field, the shape the `AgentEvent` enum already
//! declares via its `serde` derives.
//!
//! [`AgentEvent::from_host_payload`] replaces that with a typed
//! `serde_json::from_value::<AgentEvent>` path. The payload keys already
//! match the enum's snake_case field names, so most event types
//! deserialize directly once the `type` tag and `session_id` are injected.
//! Only three classes of arm need bespoke handling:
//!
//! 1. **Special arms** ([`from_host_special`]) where the host `event_type`
//!    does not map 1:1 onto a variant's fields — the whole payload becomes
//!    one field (`loop_stuck`, `cache_hit`, …), or a nudge `event_type`
//!    collapses onto a synthesized `FeedbackInjected`.
//! 2. **Field defaults** ([`apply_host_payload_defaults`]) for the handful
//!    of genuinely-optional payload fields the old match defaulted to a
//!    non-serde-default value (`ToolCall.status` → `pending`,
//!    `progress_reported.replace` → `true`, the container fields that
//!    default to `[]`/`{}`, …), plus the bare-string `executor` alias
//!    normalization the internally-tagged [`super::ToolExecutor`] can't
//!    parse on its own. Required scalars the loop always emits are left to
//!    serde (a malformed emit surfaces a loud error instead of a silent
//!    zero-fill).
//! 3. **Ambient audit** — `tool_call` / `tool_call_update` take their
//!    `audit` from the active mutation session, never the payload.
//!
//! [`HOST_EVENT_POLICIES`] is the single registry for this boundary. It owns
//! both which `event_type` strings may enter through the host path and whether
//! each accepted event is copied into the live transcript journal. Many
//! `AgentEvent` variants (`worker_update`, `handoff`, `artifact`, …) are
//! constructed elsewhere and are *not* emittable through this host path.

use std::collections::BTreeMap;

use serde_json::{Map, Value};

use crate::value::VmError;

use super::AgentEvent;

const HOST_AGENT_EMIT_EVENT: &str = "__host_agent_emit_event";
const NO_PROGRESS_STREAK_NUDGE_FALLBACK: &str =
    "No progress was detected. Use the next turn to make concrete task progress or explain the remaining blocker.";

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum HostTranscriptRole {
    Assistant,
    Tool,
}

impl HostTranscriptRole {
    pub(crate) const fn as_str(self) -> &'static str {
        match self {
            Self::Assistant => "assistant",
            Self::Tool => "tool",
        }
    }
}

/// How much of a host payload the arm behind a registry row actually reads.
///
/// "Declared as a field" is not "consumed": some arms fold a payload key into
/// a differently-named field, and some take the whole payload as one value.
/// The census in [`dropped_payload_keys`] needs that distinction to tell a
/// key nobody reads from a key that is read under another name.
#[derive(Clone, Copy, Debug)]
enum PayloadConsumption {
    /// The arm keeps the payload whole, so no key can be lost.
    Whole,
    /// Every consumed key survives into the serialized event under its own
    /// name, so a round trip decides.
    Fields,
    /// A round trip decides, except for these keys, which the arm reads and
    /// folds into a differently-named field.
    FieldsFolding(&'static [&'static str]),
}

#[derive(Clone, Copy, Debug)]
struct HostEventPolicy {
    event_type: &'static str,
    transcript_role: Option<HostTranscriptRole>,
    payload_consumption: PayloadConsumption,
}

/// A row whose arm keeps every consumed key under its own name.
///
/// This is the default on purpose. A row added without thinking about
/// consumption over-reports a fold as a drop, which is noisy and visible; the
/// opposite default would silently exempt the next arm from the census and
/// rebuild the very defect this registry exists to expose.
const fn host_event(
    event_type: &'static str,
    transcript_role: Option<HostTranscriptRole>,
) -> HostEventPolicy {
    HostEventPolicy {
        event_type,
        transcript_role,
        payload_consumption: PayloadConsumption::Fields,
    }
}

/// A row whose arm stores the payload whole (`checkpoint`, `payload`, …).
const fn host_event_whole(
    event_type: &'static str,
    transcript_role: Option<HostTranscriptRole>,
) -> HostEventPolicy {
    HostEventPolicy {
        event_type,
        transcript_role,
        payload_consumption: PayloadConsumption::Whole,
    }
}

/// A row whose arm reads `folded` keys into differently-named fields.
const fn host_event_folding(
    event_type: &'static str,
    transcript_role: Option<HostTranscriptRole>,
    folded: &'static [&'static str],
) -> HostEventPolicy {
    HostEventPolicy {
        event_type,
        transcript_role,
        payload_consumption: PayloadConsumption::FieldsFolding(folded),
    }
}

const ASSISTANT: Option<HostTranscriptRole> = Some(HostTranscriptRole::Assistant);
const TOOL: Option<HostTranscriptRole> = Some(HostTranscriptRole::Tool);

/// The one policy registry for events entering through `agent_emit_event`.
///
/// A registry row authorizes host deserialization. `transcript_role` controls
/// whether the same accepted payload is copied into the durable live-session
/// journal. Keeping both decisions together prevents an event from appearing
/// registered at its stdlib call site while being silently absent from one of
/// the runtime's observation surfaces.
const HOST_EVENT_POLICIES: &[HostEventPolicy] = &[
    host_event("tool_call", ASSISTANT),
    host_event("tool_call_update", ASSISTANT),
    host_event("iteration_start", None),
    host_event("iteration_end", None),
    host_event("judge_started", None),
    host_event("judge_decision", None),
    host_event("step_judge_decision", None),
    host_event("structural_validator_decision", None),
    host_event("scope_classifier_verdict", None),
    host_event("input_guardrail_verdict", None),
    host_event("missing_tool_call_verdict", None),
    host_event("require_successful_tools_violation", ASSISTANT),
    host_event("final_wrapup", ASSISTANT),
    host_event("pack_thinking_stripped", ASSISTANT),
    host_event("self_consistency_tie", ASSISTANT),
    host_event("code_librarian_query_nl_fallback", ASSISTANT),
    host_event("budget_exhausted", ASSISTANT),
    host_event("budget_circuit_breaker", ASSISTANT),
    // Progress is user-facing product state. Journal it as an internal audit
    // event so reconnect/restart can project it without adding a provider-visible
    // message.
    host_event("progress_reported", ASSISTANT),
    // A purpose label is user-facing product state with no provider-visible
    // effect. Journalling it lets a reconnect or replay redraw the same
    // headings the live client saw.
    host_event("purpose_label", ASSISTANT),
    host_event("tool_search_query", ASSISTANT),
    host_event("tool_search_result", TOOL),
    host_event("skill_narrow", ASSISTANT),
    host_event("loop_control_decision", None),
    host_event("capability_gap", None),
    host_event("tool_format_override", ASSISTANT),
    host_event("tool_call_audit", TOOL),
    host_event("tool_batch_disposition", TOOL),
    host_event("loop_checkpoint", ASSISTANT),
    // The loud-boundary funnel (harn#5142). Registered so a `.harn` boundary
    // reports a drop through the same typed event as the Rust funnel.
    host_event("boundary_failure", None),
    host_event_whole("typed_checkpoint", ASSISTANT),
    host_event_whole("model_job", TOOL),
    host_event_whole("loop_stuck", ASSISTANT),
    host_event_whole("reserved_terminal_verify", ASSISTANT),
    host_event_whole("agent_loop_stall_warning", ASSISTANT),
    host_event_whole("cache_hit", None),
    host_event_whole("cache_miss", None),
    // `std/llm` handler telemetry. Every one of these shipped in the embedded
    // stdlib while this registry refused it, so the events were emitted and
    // dropped; the drift check in `from_host_tests` is what keeps the two
    // halves together from here.
    host_event_whole("llm_call_log", None),
    host_event_whole("llm_routing_decision", None),
    host_event_whole("llm_fallback_attempt", None),
    host_event_whole("llm_shadow_diff", None),
    host_event_whole("semantic_cache_hit", None),
    host_event_whole("semantic_cache_miss", None),
    host_event_whole("agent_scratchpad_reorganization", None),
    host_event("stance_armed", None),
    host_event("stance_write_access_granted", None),
    host_event("stance_write_access_denied", None),
    host_event("stance_disarmed", None),
    host_event_folding(
        "completion_confirmation_nudge",
        None,
        &["message", "text", "visible_text_prefix"],
    ),
    host_event_folding(
        "fenced_call_attempt_nudge",
        None,
        &["message", "text", "fence"],
    ),
    host_event_folding(
        "malformed_call_markup_nudge",
        None,
        &["message", "text", "marker"],
    ),
    host_event_folding(
        "missing_tool_call_nudge",
        None,
        &["message", "text", "tool"],
    ),
    host_event("repair_output_contract_applied", None),
    host_event_folding(
        "no_progress_streak_nudge",
        None,
        &["message", "text", "turns_since_progress"],
    ),
    host_event_folding(
        "tool_call_blank_name_dropped",
        None,
        &["message", "text", "dropped_count"],
    ),
    host_event_folding(
        "llm_auto_continue",
        None,
        &[
            "message",
            "text",
            "previous_max_tokens",
            "raised_max_tokens",
            "attempt",
            "max_continuations",
        ],
    ),
    host_event_folding(
        "context_overflow_recovery",
        ASSISTANT,
        &[
            "message",
            "text",
            "attempt",
            "max_recoveries",
            "archived_messages",
        ],
    ),
];

/// Every `event_type` this boundary accepts, for the drift check that keeps
/// the embedded stdlib's emitters and this registry in sync.
#[cfg(test)]
pub(super) fn registered_host_event_types() -> impl Iterator<Item = &'static str> {
    HOST_EVENT_POLICIES.iter().map(|policy| policy.event_type)
}

fn warn_unknown_host_event_once(session_id: &str, event_type: &str) {
    let first = crate::agent_sessions::mark_unknown_host_event_warning(session_id, event_type);
    if first {
        crate::events::log_warn_meta(
            "host_event_ingest",
            &format!("unsupported event type `{event_type}`"),
            BTreeMap::from([
                (
                    "session_id".to_string(),
                    serde_json::Value::String(session_id.to_string()),
                ),
                (
                    "event_type".to_string(),
                    serde_json::Value::String(event_type.to_string()),
                ),
            ]),
        );
    }
}

fn host_event_policy(event_type: &str) -> Option<&'static HostEventPolicy> {
    HOST_EVENT_POLICIES
        .iter()
        .find(|policy| policy.event_type == event_type)
}

/// Payload keys the emitter wrote that no reader of the decoded event can see.
///
/// `AgentEvent` is an internally-tagged enum with no `deny_unknown_fields`, so
/// `serde` discards a key the target variant does not declare. The discard is
/// the whole defect: the emit succeeds, the run succeeds, and the field is
/// missing only for whoever reads the timeline afterwards.
///
/// The census is a round trip rather than a hand-written key list. Serializing
/// the decoded event answers "what will a reader actually see" from the same
/// derives that did the dropping, so it cannot drift out of date the way a
/// second copy of every variant's fields would. The registry supplies only
/// what a round trip cannot know: which arms keep the payload whole, and which
/// keys are read under a different name.
///
/// A `null` input value is not reported. Every optional field skips
/// serializing when absent, so an explicit `null` round-trips to nothing
/// through a field that does exist and is not evidence of a drop.
fn dropped_payload_keys(
    policy: &HostEventPolicy,
    payload: &Value,
    event: &AgentEvent,
) -> Vec<String> {
    let folded: &[&str] = match policy.payload_consumption {
        PayloadConsumption::Whole => return Vec::new(),
        PayloadConsumption::Fields => &[],
        PayloadConsumption::FieldsFolding(keys) => keys,
    };
    let Value::Object(sent) = payload else {
        return Vec::new();
    };
    let Ok(Value::Object(read)) = serde_json::to_value(event) else {
        return Vec::new();
    };
    sent.iter()
        .filter(|(key, value)| {
            !value.is_null() && !read.contains_key(*key) && !folded.contains(&key.as_str())
        })
        .map(|(key, _)| key.clone())
        .collect()
}

/// Report each dropped key once per session, through the loud-boundary funnel.
///
/// [`crate::boundary::BoundaryFailureKind::Dropped`] already names this exact
/// shape — bytes consumed that produced neither action nor error — so the
/// signal joins the boundary vocabulary that exists rather than inventing a
/// parallel one. It is deliberately not an error: the population of stray keys
/// is known to be non-empty and some of it is load-bearing on live paths, so
/// refusing the event here would turn an invisible loss into a dead run.
fn report_dropped_payload_keys(
    session_id: &str,
    event_type: &str,
    policy: &HostEventPolicy,
    payload: &Value,
    event: &AgentEvent,
) {
    for key in dropped_payload_keys(policy, payload, event) {
        if !crate::agent_sessions::mark_dropped_host_payload_key_warning(
            session_id, event_type, &key,
        ) {
            continue;
        }
        crate::boundary::BoundaryFailure::new(
            crate::boundary::BoundaryId::HostEventIngest,
            crate::boundary::BoundaryFailureKind::Dropped,
            format!("`{event_type}` payload key `{key}` is not read by any field of the event it becomes"),
        )
        .in_session(session_id)
        .with_excerpt(&payload.to_string())
        .report();
    }
}

impl AgentEvent {
    /// Build a typed [`AgentEvent`] from a host `emit_event` call.
    ///
    /// Known types keep the retired `build_agent_event` accept/reject
    /// boundary: a malformed payload of a registered name is a contract
    /// violation and returns a `Runtime` error. An `event_type` this
    /// registry has never seen is not a contract violation — Harn's
    /// vocabulary is additive across pins — so the call returns `Ok(None)`
    /// after at most one warning per type per session. Callers must not
    /// journal or emit the dropped name.
    ///
    /// A registered type whose payload carries a key no field of the resulting
    /// event reads is accepted, and the lost key is reported once per session
    /// through the loud-boundary funnel. It is a report rather than a refusal
    /// because live emitters are known to pass keys nothing consumes; see
    /// [`dropped_payload_keys`].
    pub fn from_host_payload(
        session_id: &str,
        event_type: &str,
        payload: &Value,
    ) -> Result<Option<AgentEvent>, VmError> {
        let Some(policy) = host_event_policy(event_type) else {
            warn_unknown_host_event_once(session_id, event_type);
            return Ok(None);
        };
        let event = match from_host_special(session_id, event_type, payload) {
            Some(event) => event,
            None => from_host_generic(session_id, event_type, payload)?,
        };
        report_dropped_payload_keys(session_id, event_type, policy, payload, &event);
        Ok(Some(event))
    }

    pub(crate) fn host_transcript_role(event_type: &str) -> Option<HostTranscriptRole> {
        host_event_policy(event_type).and_then(|policy| policy.transcript_role)
    }
}

/// Arms whose host payload is not a 1:1 field mapping onto the variant:
/// the whole payload becomes a single field, a couple of fields are
/// derived, or a nudge `event_type` collapses onto `FeedbackInjected`.
/// Returns `None` for everything else so the caller falls through to the
/// generic deserialize path.
fn from_host_special(session_id: &str, event_type: &str, payload: &Value) -> Option<AgentEvent> {
    let sid = || session_id.to_string();
    let feedback = |kind: &str, content: String| AgentEvent::FeedbackInjected {
        session_id: sid(),
        kind: kind.to_string(),
        content,
        streak: None,
        iteration: None,
        tool_name: None,
        turn_claimed_for_repair: None,
        delivered: obj_opt_bool(payload, "delivered"),
    };
    let feedback_with_streak =
        |kind: &str, content: String, streak: Option<usize>| AgentEvent::FeedbackInjected {
            session_id: sid(),
            kind: kind.to_string(),
            content,
            streak,
            // The payload has always carried this; the projection dropped it,
            // so a receipt could not be tied to the turn that caused it without
            // parsing prose. `repair_feedback` below already read it.
            iteration: obj_opt_usize(payload, "iteration"),
            tool_name: None,
            turn_claimed_for_repair: None,
            // Absent unless the mechanism says. A receipts-only mechanism says
            // `false`; one that still injects says `true`; one that has not been
            // taught the question stays `None` rather than claiming either.
            delivered: obj_opt_bool(payload, "delivered"),
        };
    let repair_feedback =
        |kind: &str, content: String, tool_name: Option<String>| AgentEvent::FeedbackInjected {
            session_id: sid(),
            kind: kind.to_string(),
            content,
            streak: None,
            iteration: Some(obj_usize(payload, "iteration")),
            tool_name,
            turn_claimed_for_repair: Some(true),
            delivered: obj_opt_bool(payload, "delivered"),
        };
    let feedback_content = |fallback: String| {
        first_non_empty_string(payload, &["content", "message", "text"]).unwrap_or(fallback)
    };
    let event = match event_type {
        "typed_checkpoint" => AgentEvent::TypedCheckpoint {
            session_id: sid(),
            checkpoint: payload.clone(),
        },
        "model_job" => AgentEvent::ModelJob {
            session_id: sid(),
            event: payload.clone(),
        },
        "loop_stuck" => AgentEvent::LoopStuckSignal {
            session_id: sid(),
            payload: payload.clone(),
        },
        "reserved_terminal_verify" => AgentEvent::ReservedTerminalVerify {
            session_id: sid(),
            payload: payload.clone(),
        },
        "agent_loop_stall_warning" => AgentEvent::AgentLoopStallWarning {
            session_id: sid(),
            warning: payload.clone(),
        },
        "cache_hit" => AgentEvent::CacheHit {
            session_id: sid(),
            key: obj_string(payload, "key"),
            backend: obj_string(payload, "backend"),
            namespace: obj_string(payload, "namespace"),
            payload: payload.clone(),
        },
        "cache_miss" => AgentEvent::CacheMiss {
            session_id: sid(),
            key: obj_string(payload, "key"),
            backend: obj_string(payload, "backend"),
            namespace: obj_string(payload, "namespace"),
            payload: payload.clone(),
        },
        "llm_call_log" => AgentEvent::LlmCallLog {
            session_id: sid(),
            model: obj_string(payload, "model"),
            provider: obj_string(payload, "provider"),
            status: obj_string(payload, "status"),
            latency_ms: obj_usize(payload, "latency_ms"),
            iteration: obj_usize(payload, "iteration"),
            attempt: obj_usize(payload, "attempt"),
            payload: payload.clone(),
        },
        "llm_routing_decision" => AgentEvent::LlmRoutingDecision {
            session_id: sid(),
            route_index: obj_i64(payload, "route_index"),
            route_name: obj_string(payload, "route_name"),
            used_default: obj_bool(payload, "used_default"),
            payload: payload.clone(),
        },
        "llm_fallback_attempt" => AgentEvent::LlmFallbackAttempt {
            session_id: sid(),
            fallback_index: obj_usize(payload, "fallback_index"),
            fallback_total: obj_usize(payload, "fallback_total"),
            ok: obj_bool(payload, "ok"),
            status: obj_string(payload, "status"),
            payload: payload.clone(),
        },
        "llm_shadow_diff" => AgentEvent::LlmShadowDiff {
            session_id: sid(),
            primary_ok: obj_bool(payload, "primary_ok"),
            shadow_ok: obj_bool(payload, "shadow_ok"),
            primary_status: obj_string(payload, "primary_status"),
            shadow_status: obj_string(payload, "shadow_status"),
            primary_len: obj_usize(payload, "primary_len"),
            shadow_len: obj_usize(payload, "shadow_len"),
            payload: payload.clone(),
        },
        "semantic_cache_hit" => AgentEvent::SemanticCacheHit {
            session_id: sid(),
            similarity: obj_f64(payload, "similarity"),
            provider: obj_string(payload, "provider"),
            model: obj_string(payload, "model"),
            payload: payload.clone(),
        },
        "semantic_cache_miss" => AgentEvent::SemanticCacheMiss {
            session_id: sid(),
            nearest_similarity: obj_f64(payload, "nearest_similarity"),
            payload: payload.clone(),
        },
        "agent_scratchpad_reorganization" => {
            let mut details = payload.clone();
            if let Some(object) = details.as_object_mut() {
                object.remove("iteration");
                object.remove("status");
            }
            AgentEvent::AgentScratchpadReorganization {
                session_id: sid(),
                iteration: obj_usize(payload, "iteration"),
                status: obj_string(payload, "status"),
                details,
            }
        }
        // Read-only stance lifecycle (std/agent/stance). The four stdlib
        // event names map onto one typed variant distinguished by `phase`
        // so trace consumers match on a single event type.
        "stance_armed"
        | "stance_write_access_granted"
        | "stance_write_access_denied"
        | "stance_disarmed" => {
            let allowed_tools = payload
                .get("allowed_tools")
                .and_then(Value::as_array)
                .map(|values| {
                    values
                        .iter()
                        .filter_map(|value| value.as_str().map(str::to_string))
                        .collect()
                })
                .unwrap_or_default();
            AgentEvent::StanceTransition {
                session_id: sid(),
                phase: event_type
                    .strip_prefix("stance_")
                    .unwrap_or(event_type)
                    .to_string(),
                escape_tool: obj_string(payload, "escape_tool"),
                allowed_tools,
                justification: obj_string(payload, "justification"),
                consent: obj_string(payload, "consent"),
                reason: obj_string(payload, "reason"),
            }
        }
        // Engine-side corrective nudges (see the retired match's doc
        // comments): each surfaces to operators on the FeedbackInjected
        // stream with a synthesized `kind` and a derived `content`.
        "completion_confirmation_nudge" => feedback(
            "completion_confirmation_nudge",
            feedback_content(obj_string(payload, "visible_text_prefix")),
        ),
        "fenced_call_attempt_nudge" => repair_feedback(
            "fenced_call_attempt_nudge",
            feedback_content(obj_string(payload, "fence")),
            None,
        ),
        "malformed_call_markup_nudge" => repair_feedback(
            "malformed_call_markup_nudge",
            feedback_content(obj_string(payload, "marker")),
            None,
        ),
        "missing_tool_call_nudge" => repair_feedback(
            "missing_tool_call_nudge",
            feedback_content(obj_string(payload, "tool")),
            first_non_empty_string(payload, &["tool"]),
        ),
        "no_progress_streak_nudge" => feedback_with_streak(
            "no_progress_streak_nudge",
            feedback_content(NO_PROGRESS_STREAK_NUDGE_FALLBACK.to_string()),
            feedback_streak(payload),
        ),
        "tool_call_blank_name_dropped" => feedback(
            "tool_call_blank_name_dropped",
            feedback_content(obj_usize(payload, "dropped_count").to_string()),
        ),
        "llm_auto_continue" => feedback(
            "llm_auto_continue",
            feedback_content(format!(
                "{}->{} (attempt {}/{})",
                obj_usize(payload, "previous_max_tokens"),
                obj_usize(payload, "raised_max_tokens"),
                obj_usize(payload, "attempt"),
                obj_usize(payload, "max_continuations"),
            )),
        ),
        "context_overflow_recovery" => feedback(
            "context_overflow_recovery",
            feedback_content(format!(
                "attempt {}/{} archived {} messages",
                obj_usize(payload, "attempt"),
                obj_usize(payload, "max_recoveries"),
                obj_usize(payload, "archived_messages"),
            )),
        ),
        _ => return None,
    };
    Some(event)
}

fn first_non_empty_string(payload: &Value, keys: &[&str]) -> Option<String> {
    keys.iter().find_map(|key| {
        payload
            .get(*key)
            .and_then(Value::as_str)
            .filter(|value| !value.trim().is_empty())
            .map(str::to_string)
    })
}

fn feedback_streak(payload: &Value) -> Option<usize> {
    let streak = obj_usize(payload, "streak").max(obj_usize(payload, "turns_since_progress"));
    (streak > 0).then_some(streak)
}

/// Generic path: allowlist-check, normalize the payload to match the
/// enum's serde shape, deserialize, then override the ambient `audit`
/// for the two tool-call variants.
fn from_host_generic(
    session_id: &str,
    event_type: &str,
    payload: &Value,
) -> Result<AgentEvent, VmError> {
    let mut obj = match payload {
        Value::Object(map) => map.clone(),
        _ => Map::new(),
    };
    apply_host_payload_defaults(event_type, &mut obj)?;
    obj.insert("type".to_string(), Value::String(event_type.to_string()));
    obj.insert(
        "session_id".to_string(),
        Value::String(session_id.to_string()),
    );
    let mut event: AgentEvent = serde_json::from_value(Value::Object(obj)).map_err(|error| {
        reject(
            session_id,
            format!("invalid `{event_type}` payload: {error}"),
            payload,
        )
    })?;
    // `tool_call` / `tool_call_update` carry the mutation-session audit
    // context active at emit time, never a payload-supplied value.
    if let AgentEvent::ToolCall { audit, .. } | AgentEvent::ToolCallUpdate { audit, .. } =
        &mut event
    {
        *audit = crate::orchestration::current_mutation_session();
    }
    Ok(event)
}

/// Refuse a malformed payload of a *registered* host event, loudly.
///
/// Unknown `event_type` names are not a contract violation and do not go
/// through this funnel: they warn once per session and return `Ok(None)`.
/// A known type with an invalid payload used to vanish because every
/// stdlib emit site wraps `agent_emit_event` in `try { }` and discards the
/// result. The rejection now also goes out through the loud-boundary
/// funnel (harn#5142), which is not swallowable by a caller's `try`.
fn reject(session_id: &str, detail: String, payload: &Value) -> VmError {
    crate::boundary::BoundaryFailure::new(
        crate::boundary::BoundaryId::HostEventIngest,
        crate::boundary::BoundaryFailureKind::Unrecognized,
        detail.clone(),
    )
    .in_session(session_id)
    .with_excerpt(&payload.to_string())
    .report();
    VmError::Runtime(format!("{HOST_AGENT_EMIT_EVENT}: {detail}"))
}

/// Fill in the required-field defaults the retired hand-match applied that
/// differ from serde's own missing-field behavior (serde already defaults
/// missing `Option<T>` fields to `None`, so only non-`Option` required
/// fields with a non-zero/non-empty default need help here).
fn apply_host_payload_defaults(
    event_type: &str,
    obj: &mut Map<String, Value>,
) -> Result<(), VmError> {
    match event_type {
        "tool_call" => {
            obj.remove("audit"); // sourced from the ambient mutation session
            set_default(obj, "status", Value::String("pending".to_string()));
            set_default(obj, "raw_input", Value::Null);
        }
        "tool_call_update" => {
            obj.remove("audit"); // sourced from the ambient mutation session
            set_default(obj, "status", Value::String("in_progress".to_string()));
            normalize_executor(obj)?;
        }
        "iteration_end" => set_default(obj, "iteration_info", Value::Null),
        "progress_reported" => {
            set_default(obj, "entries", Value::Array(Vec::new()));
            set_default(obj, "replace", Value::Bool(true));
            set_default(obj, "metadata", Value::Object(Map::new()));
        }
        "purpose_label" => {
            set_default(obj, "source", Value::String("declared".to_string()));
            set_default(obj, "tool_call_ids", Value::Array(Vec::new()));
        }
        "tool_search_query" => set_default(obj, "query", Value::Null),
        "tool_search_result" => set_default(obj, "promoted", Value::Array(Vec::new())),
        "skill_narrow" => {
            set_default(obj, "removed_tools", Value::Array(Vec::new()));
            set_default(obj, "remaining_tools", Value::Array(Vec::new()));
        }
        "tool_call_audit" => set_default(obj, "audit", Value::Null),
        // `owner` is derived from `kind`, never supplied: one attribution rule
        // for the Rust funnel and the `.harn` boundaries alike. A payload that
        // tries to set it is overruled rather than trusted.
        "boundary_failure" => {
            let owner = obj
                .get("kind")
                .and_then(Value::as_str)
                .and_then(|kind| {
                    serde_json::from_value::<crate::boundary::BoundaryFailureKind>(Value::String(
                        kind.to_string(),
                    ))
                    .ok()
                })
                .map(|kind| kind.owner())
                .unwrap_or("harness");
            obj.insert("owner".to_string(), Value::String(owner.to_string()));
        }
        _ => {}
    }
    Ok(())
}

/// Normalize a bare-string `executor` into the object form
/// [`super::ToolExecutor`]'s internally-tagged `Deserialize` expects,
/// preserving the retired match's alias set. Non-string values (absent,
/// `null`, or an already-structured `mcp_server` object) are left for
/// serde to handle.
fn normalize_executor(obj: &mut Map<String, Value>) -> Result<(), VmError> {
    let raw = match obj.get("executor") {
        Some(Value::String(value)) => value.clone(),
        _ => return Ok(()),
    };
    let kind = match raw.trim() {
        "" => {
            obj.remove("executor");
            return Ok(());
        }
        "harn" | "harn_builtin" => "harn_builtin",
        "host" | "host_bridge" => "host_bridge",
        "provider" | "provider_native" => "provider_native",
        other => {
            return Err(VmError::Runtime(format!(
                "{HOST_AGENT_EMIT_EVENT}: invalid tool executor `{other}`"
            )));
        }
    };
    let mut executor = Map::new();
    executor.insert("kind".to_string(), Value::String(kind.to_string()));
    obj.insert("executor".to_string(), Value::Object(executor));
    Ok(())
}

fn set_default(obj: &mut Map<String, Value>, key: &str, value: Value) {
    obj.entry(key).or_insert(value);
}

fn obj_string(payload: &Value, key: &str) -> String {
    payload
        .get(key)
        .and_then(Value::as_str)
        .unwrap_or("")
        .to_string()
}

fn obj_usize(payload: &Value, key: &str) -> usize {
    payload.get(key).and_then(Value::as_u64).unwrap_or(0) as usize
}

/// A count the payload may simply not carry.
///
/// `obj_usize` answers 0 for an absent key, which is the right default for a
/// counter and the wrong one for a correlator: iteration 0 is a real turn, so a
/// row that never knew its iteration would claim the first one.
fn obj_opt_usize(payload: &Value, key: &str) -> Option<usize> {
    payload
        .get(key)
        .and_then(Value::as_u64)
        .map(|value| value as usize)
}

/// A yes/no the payload may simply not answer, kept apart from an explicit no.
fn obj_opt_bool(payload: &Value, key: &str) -> Option<bool> {
    payload.get(key).and_then(Value::as_bool)
}

fn obj_bool(payload: &Value, key: &str) -> bool {
    payload.get(key).and_then(Value::as_bool).unwrap_or(false)
}

fn obj_f64(payload: &Value, key: &str) -> f64 {
    payload.get(key).and_then(Value::as_f64).unwrap_or(0.0)
}

/// A route index is `-1` when the router fell through to its default, so this
/// one cannot borrow [`obj_usize`].
fn obj_i64(payload: &Value, key: &str) -> i64 {
    payload.get(key).and_then(Value::as_i64).unwrap_or(0)
}