harn-vm 0.10.153

//! Durable assistant-message history and orphaned native tool-use repair.

use super::*;

/// Return the visible message list for an agent session.
#[harn_builtin(
    exposure = "runtime_internal",
    effects = [],
    sig = "__host_agent_session_messages(session_id: string) -> list",
    category = "agent.host",
    runtime_only = true
)]
fn host_agent_session_messages_builtin(
    args: &[VmValue],
    _out: &mut String,
) -> Result<VmValue, VmError> {
    let session_id = args.first().map(|v| v.display()).unwrap_or_default();
    let snapshot = crate::agent_sessions::transcript(&session_id);
    let messages = snapshot
        .as_ref()
        .and_then(|v| dict_get(v, "messages"))
        .cloned()
        .unwrap_or_else(|| VmValue::List(std::sync::Arc::new(Vec::new())));
    Ok(messages)
}

pub(super) fn assistant_message_from_llm_result(llm_result: &VmValue) -> VmValue {
    let text = dict_get(llm_result, "text")
        .map(|v| v.display())
        .unwrap_or_default();
    let provider = dict_get(llm_result, "provider")
        .map(|v| v.display())
        .unwrap_or_default();
    let model = dict_get(llm_result, "model")
        .map(|v| v.display())
        .unwrap_or_default();
    // Only attach provider-native tool calls to the assistant envelope.
    // Text-mode calls remain inline in `text` and are parsed from there.
    let native_calls_value = dict_get(llm_result, "native_tool_calls")
        .cloned()
        .unwrap_or(VmValue::Nil);
    let native_calls_json = list_items(&native_calls_value)
        .iter()
        .map(vm_to_json)
        .collect::<Vec<_>>();
    let durable_blocks = durable_anthropic_blocks(llm_result, &provider, &model);
    let thinking = dict_get(llm_result, "thinking").map(|v| v.display());
    if let Some(message) = assistant_messages::text_channel_provider_surprise_message(
        llm_result,
        &provider,
        &model,
        &text,
        &native_calls_json,
        thinking.as_deref(),
    ) {
        return crate::llm::pairing_receipts::attach_assistant_facts(message, llm_result);
    }
    if native_calls_json.is_empty() {
        // gpt-oss / harmony channel-leak backstop. A native-tools model is
        // supposed to split its harmony channels at the wire: analysis ->
        // `reasoning`, commentary/tool -> `tool_calls`, final -> `content`. On
        // ~23% of gpt-oss-120b turns the provider FAILS to split and collapses
        // the analysis reasoning AND the inline tool-call JSON into a single
        // `content` blob (empty `reasoning` field, empty `tool_calls`). The
        // tagged-parser merge in `vm_build_llm_result` recovers the call into
        // the unified `tool_calls` (the `tool`-key dialect now recovers too,
        // see native_json.rs) and suppresses action-only wrapper text from the
        // public text/prose fields. Replaying that raw blob back into history
        // would waste input tokens AND re-feed the model its own private
        // chain-of-thought (incl. "game the verifier" plans) on every later
        // turn.
        //
        // For a native-tools model the canonical persisted shape is structured
        // `tool_calls` + a private `reasoning` trace + a clean `content` (this
        // is exactly what a NON-leaked gpt-oss turn produces, and what the
        // native-calls-present branch below builds). So we reconstruct that
        // shape: move the leaked blob into the private `reasoning` field (it is
        // analysis CoT, not a committed answer — clean tool-call turns carry no
        // `content`), attach the recovered call to `tool_calls`, and leave
        // `content` empty. The next request's openai-compat wire already strips
        // prior-turn `reasoning` (harn#3319), so nothing dirty is re-fed.
        //
        // Pure text-format models (`native_tools == false`, e.g. local
        // llamacpp) legitimately keep their calls inline in `content` for the
        // NEXT turn's text parser to re-read, so those keep the verbatim-text
        // path below.
        if assistant_messages::supports_native_history(llm_result, &provider, &model) {
            let recovered_calls = list_items(
                &dict_get(llm_result, "tool_calls")
                    .cloned()
                    .unwrap_or(VmValue::Nil),
            )
            .iter()
            .map(vm_to_json)
            .collect::<Vec<_>>();
            if !recovered_calls.is_empty() {
                // A call was recovered from dirty content. Keep content empty,
                // matching a clean native-tool turn; preserve only an explicit
                // wire reasoning field when the provider supplied one.
                let reasoning = thinking.as_deref().filter(|t| !t.is_empty());
                let msg = build_assistant_response_message("", &[], &recovered_calls, reasoning);
                return crate::llm::pairing_receipts::attach_assistant_facts(
                    json_to_vm(&msg),
                    llm_result,
                );
            }
        }
        let mut msg = crate::value::DictMap::new();
        if !durable_blocks.is_empty() {
            let message =
                build_assistant_response_message(&text, &durable_blocks, &[], thinking.as_deref());
            return crate::llm::pairing_receipts::attach_assistant_facts(
                json_to_vm(&message),
                llm_result,
            );
        }
        msg.put_str("role", "assistant");
        msg.put_str("content", text);
        return crate::llm::pairing_receipts::attach_assistant_facts(
            VmValue::dict(msg),
            llm_result,
        );
    }

    let msg = build_assistant_response_message(
        &text,
        &durable_blocks,
        &native_calls_json,
        thinking.as_deref(),
    );
    crate::llm::pairing_receipts::attach_assistant_facts(json_to_vm(&msg), llm_result)
}

/// Append the assistant turn from an llm_call result to the session log.
#[harn_builtin(
    exposure = "runtime_internal",
    effects = [],
    sig = "__host_agent_session_record_assistant(session_id: string, llm_result: dict) -> nil",
    category = "agent.host",
    runtime_only = true
)]
pub(super) fn host_agent_session_record_assistant_builtin(
    args: &[VmValue],
    _out: &mut String,
) -> Result<VmValue, VmError> {
    let session_id = args.first().map(|v| v.display()).unwrap_or_default();
    let llm_result = args.get(1).cloned().unwrap_or(VmValue::Nil);
    let provider = dict_get(&llm_result, "provider")
        .map(|v| v.display())
        .unwrap_or_default();
    let model = dict_get(&llm_result, "model")
        .map(|v| v.display())
        .unwrap_or_default();
    let effective_tool_format = dict_get(&llm_result, "_effective_tool_format")
        .map(|v| v.display())
        .filter(|format| !format.trim().is_empty());
    let raw_tool_calls = dict_get(&llm_result, "tool_calls")
        .cloned()
        .unwrap_or(VmValue::Nil);
    let calls_json = list_items(&raw_tool_calls)
        .iter()
        .map(vm_to_json)
        .collect::<Vec<_>>();
    let assistant_message = assistant_message_from_llm_result(&llm_result);
    let visible_text =
        crate::llm::agent_result_projection::visible_assistant_text(&assistant_message);
    crate::agent_sessions::inject_message(&session_id, assistant_message)
        .map_err(VmError::Runtime)?;
    if let Some(content) = visible_text {
        emit_event(&AgentEvent::AgentMessageChunk {
            session_id: session_id.clone(),
            content,
        });
    }
    assistant_messages::record_dispatch_receipt(
        &session_id,
        calls_json,
        provider,
        model,
        effective_tool_format,
    );
    Ok(VmValue::Nil)
}

/// Pop the trailing assistant turn from the session transcript. Used by
/// step_judge replace mode to discard a vetoed turn before regeneration.
/// Errors if the trailing message is not an assistant turn.
#[harn_builtin(
    exposure = "runtime_internal",
    effects = [],
    sig = "__host_agent_session_pop_last_assistant(session_id: string, replacement?: string, reason?: string) -> dict",
    category = "agent.host",
    runtime_only = true
)]
fn host_agent_session_pop_last_assistant_builtin(
    args: &[VmValue],
    _out: &mut String,
) -> Result<VmValue, VmError> {
    let session_id = args.first().map(|v| v.display()).unwrap_or_default();
    let replacement = args
        .get(1)
        .map(VmValue::display)
        .filter(|text| !text.is_empty());
    let popped =
        crate::agent_sessions::pop_last_if_assistant(&session_id).map_err(VmError::Runtime)?;
    if popped {
        if let Some(replacement) = replacement {
            crate::agent_sessions::inject_message(
                &session_id,
                bookkeeping_assistant_turn(&replacement),
            )
            .map_err(VmError::Runtime)?;
            let reason = args.get(2).map(VmValue::display).unwrap_or_default();
            crate::agent_sessions::set_last_withdrawal_reason(&session_id, &reason);
        }
    }
    Ok(VmValue::Bool(popped))
}

/// An assistant turn the loop writes for its own bookkeeping.
///
/// Built here rather than by the caller so the flag the answer projection
/// reads has exactly one writer. A caller that assembled the dict itself would
/// be a second place the key name has to stay right, and the failure mode of
/// getting it wrong is silent: the turn reads back as something the model
/// said.
fn bookkeeping_assistant_turn(text: &str) -> VmValue {
    use crate::value::VmDictExt;
    let mut message = crate::value::DictMap::new();
    message.put_str("role", "assistant");
    message.put_str("content", text);
    message.put_bool(
        crate::llm::agent_result_projection::BOOKKEEPING_TURN_KEY,
        true,
    );
    VmValue::dict(message)
}

/// True when the trailing message in the session transcript is an assistant
/// turn carrying at least one canonical native tool call (as opposed to a
/// text-channel turn that keeps calls inline in plain-string `content`). Used
/// by `record_tool_results` to keep an escalated native turn on the native
/// channel even when the primary session is text-locked.
pub(super) fn trailing_assistant_has_native_tool_use(session_id: &str) -> bool {
    let Some(transcript) = crate::agent_sessions::transcript(session_id) else {
        return false;
    };
    let messages = list_items(
        &dict_get(&transcript, "messages")
            .cloned()
            .unwrap_or(VmValue::Nil),
    );
    let Some(last) = messages.last() else {
        return false;
    };
    if dict_get(last, "role")
        .map(|v| v.display())
        .unwrap_or_default()
        != "assistant"
    {
        return false;
    }
    !assistant_tool_use_blocks(last).is_empty()
}

/// Repair the transcript invariant that every assistant native tool call is
/// followed by a matching `tool_result` before the next provider request. The
/// agent loop calls this at every inject site that
/// DECLINES to dispatch an assistant turn's tool calls (native-format fallback
/// reject, all-blank-name drop, parse-error, no-progress nudge) and would
/// otherwise append a bare user-feedback message after an orphaned `tool_use` —
/// which Anthropic rejects with a non-retryable HTTP 400 ("tool_use ids were
/// found without tool_result blocks immediately after"), killing the run.
///
/// The synthesized tool-result carries `feedback` as its observation, so the
/// model still sees the same corrective steering it would have gotten from the
/// user message — just delivered in a provider-valid tool-result envelope that
/// keeps pairing intact.
///
/// Returns the number of orphaned blocks repaired. `0` when the trailing message
/// is not an assistant turn, carries no structured tool_use (e.g. a homogeneous
/// text-format run keeps calls inline in `content`), or every block already has
/// a paired result — so this is a strict no-op for runs that already converge.
pub(super) fn pair_orphaned_tool_use(session_id: &str, feedback: &str) -> usize {
    let Some(transcript) = crate::agent_sessions::transcript(session_id) else {
        return 0;
    };
    let messages = list_items(
        &dict_get(&transcript, "messages")
            .cloned()
            .unwrap_or(VmValue::Nil),
    );
    // The assistant turn whose calls are open is the last one, with only its
    // own tool results after it. A cancel between parallel calls leaves some
    // answered and some not, so results may already trail it (harn#8951).
    let Some(last) = messages
        .iter()
        .rposition(|message| message_role(message) == "assistant")
        .filter(|index| {
            messages[index + 1..]
                .iter()
                .all(|message| is_tool_result_role(&message_role(message)))
        })
        .map(|index| &messages[index])
    else {
        return 0;
    };
    let already_paired = paired_tool_result_ids(&messages);
    let synthetic = synthesize_orphan_tool_results(last, feedback, &already_paired);
    let mut repaired = 0;
    for message in synthetic {
        if crate::agent_sessions::inject_message(session_id, message).is_ok() {
            repaired += 1;
        }
    }
    repaired
}

fn message_role(message: &VmValue) -> String {
    dict_get(message, "role")
        .map(|v| v.display())
        .unwrap_or_default()
}

fn is_tool_result_role(role: &str) -> bool {
    role == "tool_result" || role == "tool"
}

/// Call ids a message answers through content blocks rather than as a tool
/// message of its own: Anthropic `tool_result` blocks in a user turn, Gemini
/// `functionResponse` parts, and Responses `function_call_output` items.
fn block_result_ids(message: &VmValue) -> Vec<String> {
    let Some(content) = dict_get(message, "content") else {
        return Vec::new();
    };
    list_items(content)
        .iter()
        .filter_map(|block| {
            let block_type = dict_get(block, "type")
                .map(|v| v.display())
                .unwrap_or_default();
            match block_type.as_str() {
                "tool_result" => dict_get(block, "tool_use_id"),
                "function_call_output" => dict_get(block, "call_id"),
                _ => dict_get(block, "functionResponse").and_then(|part| dict_get(part, "id")),
            }
            .map(|id| id.display())
            .filter(|id| !id.is_empty())
        })
        .collect()
}

/// What the model reads for a tool call no result answered (harn#8951).
pub(crate) const UNANSWERED_TOOL_CALL_OBSERVATION: &str =
    "No result: the turn was stopped before this tool call finished.";

/// Answer every native tool call in `messages` that no tool result pairs.
///
/// Providers reject a request that carries an unanswered call: OpenAI with
/// "No tool output found for function call", Anthropic with "tool_use ids were
/// found without tool_result blocks". A turn stopped mid-tool leaves exactly
/// that call in the stored transcript, and every later request, a resumed
/// session's included, then fails the same way (harn#8951). Each unanswered
/// call gets the harness's typed repair result, placed right after the
/// assistant turn's own results so the pairing is adjacent for every provider.
///
/// Pure over its input: the stored transcript is never rewritten here, so a
/// transcript written before this fix heals on its next request. Only calls
/// with an id are answered; a call without one cannot be paired by any
/// provider, and inventing a result for it would add a message the provider
/// never asked for. Returns `None` when every call is already answered.
pub(crate) fn answer_unanswered_tool_calls(
    messages: &[serde_json::Value],
) -> Option<Vec<serde_json::Value>> {
    let vm_messages: Vec<VmValue> = messages
        .iter()
        .map(crate::stdlib::json_to_vm_value)
        .collect();
    let mut paired = paired_tool_result_ids(&vm_messages);
    paired.extend(vm_messages.iter().flat_map(block_result_ids));
    let unanswered = |message: &VmValue| {
        if message_role(message) != "assistant" {
            return Vec::new();
        }
        assistant_tool_use_blocks(message)
            .into_iter()
            .filter(|block| !block.id.is_empty() && !paired.contains(&block.id))
            .collect::<Vec<_>>()
    };
    if !vm_messages
        .iter()
        .any(|message| !unanswered(message).is_empty())
    {
        return None;
    }
    let mut out = Vec::with_capacity(messages.len() + 1);
    let mut index = 0;
    while index < messages.len() {
        out.push(messages[index].clone());
        let open = unanswered(&vm_messages[index]);
        index += 1;
        if open.is_empty() {
            continue;
        }
        while index < messages.len() && is_tool_result_role(&message_role(&vm_messages[index])) {
            out.push(messages[index].clone());
            index += 1;
        }
        for block in open {
            let answer = super::tool_result_messages::tool_result_message(
                super::tool_result_messages::ToolResultMessageInput {
                    channel: crate::llm_config::ToolFormatChannel::Native,
                    name: &block.name,
                    tool_call_id: &block.id,
                    observation: UNANSWERED_TOOL_CALL_OBSERVATION,
                    ok: false,
                    screenshots: &[],
                    data: None,
                    origin: ToolResultOrigin::HarnessRepair,
                },
            );
            out.push(crate::llm::helpers::vm_value_to_json(&answer));
        }
    }
    Some(out)
}

/// Synthesize a matching tool-result for each orphaned `tool_use`/`tool_call`
/// block on the trailing assistant turn, carrying `feedback` as the observation,
/// so a subsequent user-feedback inject never leaves the block unpaired. Returns
/// the number of blocks repaired (`0` = no-op: not an assistant turn, no
/// structured tool calls, or already paired). See `pair_orphaned_tool_use`.
#[harn_builtin(
    exposure = "runtime_internal",
    effects = [],
    sig = "__host_agent_session_pair_orphaned_tool_use(session_id: string, feedback: string) -> int",
    category = "agent.host",
    runtime_only = true
)]
fn host_agent_session_pair_orphaned_tool_use_builtin(
    args: &[VmValue],
    _out: &mut String,
) -> Result<VmValue, VmError> {
    let session_id = args.first().map(|v| v.display()).unwrap_or_default();
    let feedback = args.get(1).map(|v| v.display()).unwrap_or_default();
    let repaired = pair_orphaned_tool_use(&session_id, &feedback);
    Ok(VmValue::Int(repaired as i64))
}

const MESSAGE_HISTORY_BUILTINS: &[&VmBuiltinDef] = &[
    &HOST_AGENT_SESSION_MESSAGES_BUILTIN_DEF,
    &HOST_AGENT_SESSION_RECORD_ASSISTANT_BUILTIN_DEF,
    &HOST_AGENT_SESSION_POP_LAST_ASSISTANT_BUILTIN_DEF,
    &HOST_AGENT_SESSION_PAIR_ORPHANED_TOOL_USE_BUILTIN_DEF,
];

pub(super) fn register_message_history_primitives(vm: &mut Vm) {
    register_builtin_defs(vm, MESSAGE_HISTORY_BUILTINS);
}

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