harn-vm 0.10.156

//! Auto-compaction — transcript size management strategies.

use crate::llm::{vm_call_llm_full, vm_value_to_json};
use crate::value::{VmError, VmValue};

mod classification;
mod config;
mod policy;
mod prompt;
mod tool_output;
use crate::vm::AsyncBuiltinCtx;
pub use classification::{
    ClassificationApplicationReason, ClassificationChoice, ClassificationConfig,
    ClassificationDecision, ClassificationDecisionReceipt, ClassificationReceipt,
    ClassificationSource, ClassificationStatus,
};
pub use config::{
    compact_strategy_name, parse_compact_strategy, AutoCompactConfig, CompactStrategy,
    CompactionRequestProvenance, CompactionThresholdSource, DEFAULT_RECAP_BUDGET_BYTES,
};
pub use policy::{
    compaction_policy_metadata_fields, compaction_policy_option_keys,
    compaction_policy_to_vm_value, parse_compaction_policy_options, CompactionPolicy,
    CompactionRequest,
};
use prompt::render_llm_compaction_prompt;
pub use tool_output::{
    microcompact_tool_output, microcompact_tool_output_result, MicrocompactedToolOutput,
};

/// Observation-mask recap metrics, carried verbatim (typed) inside
/// [`super::CompactionReceipt`] so recap behavior survives every projection.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(default)]
pub struct RecapMetrics {
    pub recap_bytes: usize,
    pub budget_bytes: usize,
    pub kept_results_count: usize,
    pub dropped_count: usize,
    pub carried_prior_recap: bool,
}

impl RecapMetrics {
    pub fn to_json(self) -> serde_json::Value {
        serde_json::json!({
            "recap_bytes": self.recap_bytes,
            "budget_bytes": self.budget_bytes,
            "kept_results_count": self.kept_results_count,
            "dropped_count": self.dropped_count,
            "carried_prior_recap": self.carried_prior_recap,
        })
    }
}

/// Estimate token count from a list of JSON messages (chars / 4 heuristic).
pub fn estimate_message_tokens(messages: &[serde_json::Value]) -> usize {
    messages.iter().map(estimate_message_chars).sum::<usize>() / 4
}

fn estimate_message_chars(message: &serde_json::Value) -> usize {
    let mut total = message
        .get("content")
        .map(estimate_content_chars)
        .unwrap_or_default();
    if let Some(reasoning) = message.get("reasoning") {
        total += estimate_content_chars(reasoning);
    }
    if let Some(tool_calls) = message.get("tool_calls") {
        total += estimate_content_chars(tool_calls);
    }
    total
}

fn estimate_content_chars(value: &serde_json::Value) -> usize {
    match value {
        serde_json::Value::String(text) => text.len(),
        serde_json::Value::Array(items) => items.iter().map(estimate_content_chars).sum(),
        serde_json::Value::Object(map) => map.values().map(estimate_content_chars).sum(),
        serde_json::Value::Null => 0,
        other => other.to_string().len(),
    }
}

fn is_reasoning_or_tool_turn_message(message: &serde_json::Value) -> bool {
    let role = message
        .get("role")
        .and_then(|value| value.as_str())
        .unwrap_or_default();
    role == "tool"
        || message.get("tool_calls").is_some()
        || message
            .get("reasoning")
            .map(|value| !value.is_null())
            .unwrap_or(false)
}

fn find_prev_user_boundary(messages: &[serde_json::Value], start: usize) -> Option<usize> {
    (0..=start)
        .rev()
        .find(|idx| messages[*idx].get("role").and_then(|value| value.as_str()) == Some("user"))
}

/// True when the message carries a tool result: the OpenAI durable shape
/// (`role: "tool"`), the Anthropic durable shape (`role: "tool_result"`), or
/// a user message whose content blocks include a `tool_result`. Text-channel
/// results are ordinary user strings and intentionally don't match — they
/// have no provider-level pairing to protect.
fn is_tool_result_message(message: &serde_json::Value) -> bool {
    match message.get("role").and_then(|role| role.as_str()) {
        Some("tool") | Some("tool_result") => true,
        Some("user") => message
            .get("content")
            .and_then(|content| content.as_array())
            .is_some_and(|blocks| {
                blocks.iter().any(|block| {
                    block.get("type").and_then(|value| value.as_str()) == Some("tool_result")
                })
            }),
        _ => false,
    }
}

/// A compaction split must never land between an assistant tool-use message
/// and its tool_result message(s): a kept window that begins with a
/// tool_result whose request was drained is rejected by providers as an
/// orphaned result. Results always immediately follow their request, so a
/// split index is unsafe exactly when it points AT a tool-result message.
/// Walk backward to the message that initiated the result run (keeping the
/// request together with its results); when that would consume the whole
/// compactable window, walk forward past the run instead so compaction still
/// makes progress.
fn snap_split_off_tool_results(
    messages: &[serde_json::Value],
    split_at: usize,
    compact_start: usize,
) -> usize {
    if split_at >= messages.len() || !is_tool_result_message(&messages[split_at]) {
        return split_at;
    }
    let mut backward = split_at;
    while backward > compact_start && is_tool_result_message(&messages[backward]) {
        backward -= 1;
    }
    if backward > compact_start {
        return backward;
    }
    let mut forward = split_at;
    while forward < messages.len() && is_tool_result_message(&messages[forward]) {
        forward += 1;
    }
    forward
}

/// True when `trimmed` (an already-trimmed line) begins with `file:line` —
/// a colon immediately followed by a digit, with no whitespace before the
/// colon. Used as a strong signal that a line carries a located diagnostic.
fn line_has_file_line_prefix(trimmed: &str) -> bool {
    let bytes = trimmed.as_bytes();
    let mut i = 0;
    while i < bytes.len() && bytes[i] != b':' {
        i += 1;
    }
    i < bytes.len() && i + 1 < bytes.len() && bytes[i + 1].is_ascii_digit()
}

/// True when a single line carries failure signal worth preserving verbatim
/// when we shrink a large tool output. This is the ONE filter shared by both
/// the microcompact path (`microcompact_tool_output`) and the observation-mask
/// path (`default_mask_tool_result`). Keep these paths on one filter so
/// assertion values, rustc continuation lines, and structured failing-line
/// markers survive in the detail the model re-reads to fix the bug.
///
///   - assertion value lines with no `file:line` (`left:`, `right:`,
///     `expected:`, `actual:`, `got`, `want`) — the values the model needs to
///     see the actual-vs-expected mismatch.
///   - rustc continuation lines (`-->` location pointers, `= help:`/`= note:`,
///     numbered source rows like `12 |`, and `^^^` carets).
///   - `Lnnn:` failing-line structure some test harnesses emit.
pub(super) fn is_failure_signal_line(line: &str) -> bool {
    let trimmed = line.trim();
    if trimmed.is_empty() {
        return false;
    }
    let lower = trimmed.to_lowercase();

    let has_file_line = line_has_file_line_prefix(trimmed);
    let has_strong_keyword =
        trimmed.contains("FAIL") || trimmed.contains("panic") || trimmed.contains("Panic");
    let has_weak_keyword = trimmed.contains("error")
        || trimmed.contains("undefined")
        || trimmed.contains("expected")
        || trimmed.contains("got")
        || lower.contains("cannot find")
        || lower.contains("not found")
        || lower.contains("no such")
        || lower.contains("unresolved")
        || lower.contains("missing")
        || lower.contains("declared but not used")
        || lower.contains("unused")
        || lower.contains("mismatch");
    let positional = lower.contains(" error ")
        || lower.starts_with("error:")
        || lower.starts_with("warning:")
        || lower.starts_with("note:")
        || lower.contains("panic:");

    let assertion_value = lower.starts_with("left:")
        || lower.starts_with("right:")
        || lower.starts_with("expected:")
        || lower.starts_with("actual:")
        || lower.starts_with("got:")
        || lower.starts_with("want:")
        || lower.starts_with("got ")
        || lower.starts_with("want ")
        || lower.starts_with("assertion")
        || lower.contains("assertionerror");

    // rustc continuation lines: location pointer, help/note, numbered source
    // rows (`12 | ...`), and caret underlines (`^^^`).
    let rustc_continuation = trimmed.starts_with("-->")
        || trimmed.starts_with("= help:")
        || trimmed.starts_with("= note:")
        || trimmed.contains('^')
        || {
            // `<digits> |` numbered source row from rustc's snippet rendering.
            let mut chars = trimmed.chars();
            let mut saw_digit = false;
            let mut rest = trimmed;
            while let Some(c) = chars.clone().next() {
                if c.is_ascii_digit() {
                    saw_digit = true;
                    chars.next();
                    rest = chars.as_str();
                } else {
                    break;
                }
            }
            saw_digit && rest.trim_start().starts_with('|')
        };

    #[expect(
        clippy::string_slice,
        reason = "digits is an ASCII-digit prefix of rest, so its len is a boundary"
    )]
    let failing_line_marker = {
        if let Some(rest) = trimmed.strip_prefix('L') {
            let digits: String = rest.chars().take_while(|c| c.is_ascii_digit()).collect();
            !digits.is_empty() && rest[digits.len()..].starts_with(':')
        } else {
            false
        }
    };

    has_strong_keyword
        || (has_file_line && has_weak_keyword)
        || positional
        || assertion_value
        || rustc_continuation
        || failing_line_marker
}

/// Snap a byte offset to the nearest preceding line boundary (end of a complete line).
/// Returns the substring from the start up to and including the last complete line
/// that fits within `max_bytes`. Never cuts mid-line.
#[expect(
    clippy::string_slice,
    reason = "search_end is a floor_char_boundary; pos indexes an ASCII newline"
)]
fn snap_to_line_end(s: &str, max_bytes: usize) -> &str {
    if max_bytes >= s.len() {
        return s;
    }
    let search_end = s.floor_char_boundary(max_bytes);
    match s[..search_end].rfind('\n') {
        Some(pos) => &s[..pos + 1],
        None => &s[..search_end], // single long line — fall back to char boundary
    }
}

/// Snap a byte offset to the nearest following line boundary (start of a complete line).
/// Returns the substring from the first complete line at or after `start_byte`.
/// Never cuts mid-line.
#[expect(
    clippy::string_slice,
    reason = "search_start is a ceil_char_boundary; line_start follows an ASCII newline"
)]
fn snap_to_line_start(s: &str, start_byte: usize) -> &str {
    if start_byte == 0 {
        return s;
    }
    let search_start = s.ceil_char_boundary(start_byte);
    if search_start >= s.len() {
        return "";
    }
    match s[search_start..].find('\n') {
        Some(pos) => {
            let line_start = search_start + pos + 1;
            if line_start < s.len() {
                &s[line_start..]
            } else {
                &s[search_start..]
            }
        }
        None => &s[search_start..], // already at start of last line
    }
}

/// A compaction summary plus the size of the scaffold the summarizer generated
/// around it (headers, budget notices, drop markers).
///
/// `text.len() - scaffold_bytes` is `carried_source_bytes`: how many bytes of
/// the archived source window actually survived into the summary. A summary can
/// be hundreds of bytes long and still carry nothing — a bare
/// `[auto-compacted N older messages]` header is the shape that silently
/// destroys context — so the byte length of the summary is not a usable
/// measurement on its own.
#[derive(Debug)]
pub(crate) struct CompactionSummary {
    pub text: String,
    pub scaffold_bytes: usize,
}

impl CompactionSummary {
    fn new(text: String, scaffold_bytes: usize) -> Self {
        let scaffold_bytes = scaffold_bytes.min(text.len());
        Self {
            text,
            scaffold_bytes,
        }
    }

    pub(crate) fn carried_source_bytes(&self) -> usize {
        self.text.len().saturating_sub(self.scaffold_bytes)
    }
}

/// Typed measurement of one compaction's source window against the summary that
/// replaced it.
///
/// Every field is `Option` on purpose: `None` means the measurement was not
/// taken on this path, `Some(0)` means it was taken and read zero. Collapsing
/// the two would make an unmeasured compaction indistinguishable from one that
/// provably carried nothing forward.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(default)]
pub struct CompactionSourceMeasurement {
    /// Messages drained from the live transcript.
    pub source_message_count: Option<usize>,
    /// Plain-text content bytes in those messages, flattened across the string
    /// and block-array `content` shapes.
    pub source_bytes: Option<usize>,
    /// Bytes of the summary that replaced them.
    pub summary_bytes: Option<usize>,
    /// Bytes of that summary that came from the source window rather than from
    /// generated scaffold.
    pub carried_source_bytes: Option<usize>,
}

impl CompactionSourceMeasurement {
    /// A non-empty source window whose summary carried nothing forward is a
    /// failed compaction, never a successful one.
    pub fn discarded_source_context(&self) -> bool {
        matches!(
            (self.source_bytes, self.carried_source_bytes),
            (Some(source), Some(0)) if source > 0
        )
    }
}

/// Plain-text content bytes of one message, across the string `content` shape
/// and the block-array shape real agent transcripts carry.
fn message_content_bytes(message: &serde_json::Value) -> usize {
    message
        .get("content")
        .map(|content| super::repair_ledger::value_text(content).len())
        .unwrap_or(0)
}

fn truncate_compaction_summary(
    old_messages: &[serde_json::Value],
    archived_count: usize,
) -> CompactionSummary {
    truncate_compaction_summary_with_context(old_messages, archived_count, false)
}

fn truncate_compaction_summary_with_context(
    old_messages: &[serde_json::Value],
    archived_count: usize,
    is_llm_fallback: bool,
) -> CompactionSummary {
    let per_msg_limit = 500_usize;
    // Flatten `content` through `value_text` rather than `as_str`: a real agent
    // transcript carries tool results and multi-part turns as a block array, and
    // reading only the string shape drops every one of them — leaving a bare
    // header as the "summary" of a window that is being deleted.
    let mut carried = 0_usize;
    let summary_parts: Vec<String> = old_messages
        .iter()
        .filter_map(|m| {
            let role = m.get("role")?.as_str()?;
            let content = super::repair_ledger::value_text(m.get("content")?);
            if content.is_empty() {
                return None;
            }
            #[expect(
                clippy::string_slice,
                reason = "floor_char_boundary returns a char boundary"
            )]
            let truncated = if content.len() > per_msg_limit {
                format!(
                    "{}... [truncated from {} chars]",
                    &content[..content.floor_char_boundary(per_msg_limit)],
                    content.len()
                )
            } else {
                content
            };
            carried += truncated.len();
            Some(format!("[{role}] {truncated}"))
        })
        .take(15)
        .collect();
    let header = if is_llm_fallback {
        format!(
            "[auto-compact fallback: LLM summarizer returned empty; {archived_count} older messages abbreviated to ~{per_msg_limit} chars each]"
        )
    } else {
        format!("[auto-compacted {archived_count} older messages via truncate strategy]")
    };
    let text = format!(
        "{header}\n{}{}",
        summary_parts.join("\n"),
        if archived_count > 15 {
            format!("\n... and {} more", archived_count - 15)
        } else {
            String::new()
        }
    );
    // `take(15)` can drop parts that `carried` already counted; clamp so the
    // scaffold size can never be reported as negative work.
    let carried = carried.min(text.len());
    let scaffold_bytes = text.len() - carried;
    CompactionSummary::new(text, scaffold_bytes)
}

fn compact_summary_text_from_value(value: &VmValue) -> Result<String, VmError> {
    if let Some(map) = value.as_dict() {
        if let Some(summary) = map.get("summary").or_else(|| map.get("text")) {
            return Ok(summary.display());
        }
    }
    match value {
        VmValue::String(text) => Ok(text.to_string()),
        VmValue::Nil => Ok(String::new()),
        _ => serde_json::to_string_pretty(&vm_value_to_json(value))
            .map_err(|e| VmError::Runtime(format!("custom compactor encode error: {e}"))),
    }
}

async fn llm_compaction_summary(
    old_messages: &[serde_json::Value],
    retained_messages: &[serde_json::Value],
    archived_count: usize,
    llm_opts: &crate::llm::api::LlmCallOptions,
    summarize_prompt: Option<&str>,
    policy: &CompactionPolicy,
) -> Result<CompactionSummary, VmError> {
    let prompt = render_llm_compaction_prompt(
        summarize_prompt,
        old_messages,
        retained_messages,
        archived_count,
        policy,
    )?;
    let compact_opts = llm_opts.isolated_request(prompt, "compaction", "compact")?;
    let result = vm_call_llm_full(&compact_opts).await?;
    let summary = result.text.trim();
    if summary.is_empty() {
        // Bounded retry: one deterministic pass over the same source window.
        // The boundary still checks what that pass carried forward.
        Ok(truncate_compaction_summary_with_context(
            old_messages,
            archived_count,
            true,
        ))
    } else {
        let header = format!("[auto-compacted {archived_count} older messages]\n");
        let scaffold_bytes = header.len();
        Ok(CompactionSummary::new(
            format!("{header}{summary}"),
            scaffold_bytes,
        ))
    }
}

async fn custom_compaction_summary(
    ctx: Option<&AsyncBuiltinCtx>,
    old_messages: &[serde_json::Value],
    archived_count: usize,
    callback: &VmValue,
    reminders: &[VmValue],
    policy: &CompactionPolicy,
) -> Result<CompactionSummary, VmError> {
    let Some(VmValue::Closure(closure)) = Some(callback.clone()) else {
        return Err(VmError::Runtime(
            "compact_callback must be a closure when compact_strategy is 'custom'".to_string(),
        ));
    };
    let Some(ctx) = ctx else {
        return Err(VmError::Runtime(
            "custom transcript compaction requires an async builtin VM context".to_string(),
        ));
    };
    let mut vm = ctx.child_vm();
    let messages_vm = VmValue::List(std::sync::Arc::new(
        old_messages
            .iter()
            .map(crate::stdlib::json_to_vm_value)
            .collect(),
    ));
    let result = if policy.has_metadata()
        && (closure.func.params.len() >= 3 || closure.func.has_rest_param)
    {
        let reminders_vm = VmValue::List(std::sync::Arc::new(reminders.to_vec()));
        let policy_vm = compaction_policy_to_vm_value(policy);
        vm.call_closure_pub(&closure, &[messages_vm, reminders_vm, policy_vm])
            .await
    } else if closure.func.params.len() >= 2 || closure.func.has_rest_param {
        let reminders_vm = VmValue::List(std::sync::Arc::new(reminders.to_vec()));
        vm.call_closure_pub(&closure, &[messages_vm, reminders_vm])
            .await
    } else {
        vm.call_closure_pub(&closure, &[messages_vm]).await
    };
    let summary = compact_summary_text_from_value(&result?)?;
    ctx.forward_output(&vm.take_output());
    if summary.trim().is_empty() {
        // Bounded retry, same contract as the LLM strategy.
        Ok(truncate_compaction_summary(old_messages, archived_count))
    } else {
        let header = format!("[auto-compacted {archived_count} older messages]\n");
        let scaffold_bytes = header.len();
        Ok(CompactionSummary::new(
            format!("{header}{summary}"),
            scaffold_bytes,
        ))
    }
}

/// Marker the host emits inside a tool-output (or message) body to pin its
/// live grounding — the current file view and just-edited window — so it
/// survives a compaction pass. The host renders this literal substring inside
/// markdown headings (e.g. `## Exact current file text [no-compact]`,
/// `## Edited region now reads (...) [no-compact]`) in
/// `lib/tools/result-format.harn`. Compaction matches the substring; it does
/// not invent a new vocabulary.
pub(crate) const NO_COMPACT_MARKER: &str = "[no-compact]";

/// Upper bound on how many of the most-recent pinned segments survive a
/// compaction pass verbatim. A pin that could never be evicted would let a
/// long session accumulate unbounded pinned snapshots (e.g. one edited-window
/// per edit) and eventually overflow the context window — defeating the
/// purpose of compaction. Keeping only the latest few preserves the agent's
/// *current* grounding (the file it is editing now, emitted as the exact-text
/// block plus the numbered-lines block in one or two adjacent outputs) while
/// letting stale duplicates from earlier in the session compact normally.
pub(crate) const MAX_PINNED_SEGMENTS: usize = 3;

/// Whether a content body carries the host's `[no-compact]` pin marker.
fn is_pinned_content(content: &str) -> bool {
    content.contains(NO_COMPACT_MARKER)
}

/// Compute the set of message indices into `messages` that are pinned AND fall
/// within the most-recent [`MAX_PINNED_SEGMENTS`] pinned bodies. Older pinned
/// bodies are intentionally excluded so they compact normally (the bound).
/// `content_of` extracts the body text to inspect for each message.
fn latest_pinned_indices<'a, F>(
    messages: impl Iterator<Item = &'a serde_json::Value>,
    content_of: F,
) -> std::collections::HashSet<usize>
where
    F: Fn(&serde_json::Value) -> Option<&str>,
{
    // Walk newest-first, collecting up to MAX_PINNED_SEGMENTS pinned indices.
    let pinned: Vec<usize> = messages
        .enumerate()
        .filter(|(_, msg)| content_of(msg).is_some_and(is_pinned_content))
        .map(|(idx, _)| idx)
        .collect();
    pinned.into_iter().rev().take(MAX_PINNED_SEGMENTS).collect()
}

/// Check whether a tool-result string should be preserved verbatim during
/// observation masking. Uses content length as the primary heuristic:
/// short results (< 500 chars) are kept since they're typically error messages,
/// status lines, or concise answers that are cheap to retain and risky to mask.
/// Long results are masked to save context budget.
fn content_should_preserve(content: &str) -> bool {
    content.len() < 500
}

/// Default per-message masking for tool results.
///
/// Beyond the first-line preview, this KEEPS any failure-signal lines
/// (assertion values, located diagnostics, rustc help/caret/source rows,
/// `Lnnn:` markers) via the shared [`is_failure_signal_line`] filter, so the
/// model re-reads the actual-vs-expected detail it needs to fix the bug. The
/// previous mask dropped everything but the first line, silently shredding the
/// structured failure that the strong microcompact filter preserves at the
/// emission side.
fn default_mask_tool_result(role: &str, content: &str) -> String {
    let first_line = content.lines().next().unwrap_or(content);
    let line_count = content.lines().count();
    let char_count = content.len();
    if line_count <= 3 {
        return format!("[{role}] {content}");
    }
    #[expect(
        clippy::string_slice,
        reason = "floor_char_boundary returns a char boundary"
    )]
    let preview = &first_line[..first_line.floor_char_boundary(120)];
    // Preserve failure-signal lines (bounded so a huge log can't defeat the
    // mask). Skip the first line itself — it is already in the preview.
    let kept: Vec<&str> = content
        .lines()
        .skip(1)
        .filter(|line| is_failure_signal_line(line))
        .take(32)
        .collect();
    if kept.is_empty() {
        format!("[{role}] {preview}... [{line_count} lines, {char_count} chars masked]")
    } else {
        format!(
            "[{role}] {preview}... [{line_count} lines, {char_count} chars masked; \
             failure lines preserved]\n{}",
            kept.join("\n")
        )
    }
}

/// Stable marker on the first line of every observation-mask recap. A prior
/// recap re-enters the archive window on a later compaction as an ordinary
/// `{role: "user"}` message; matching this sentinel lets the next compaction
/// carry it forward once (bounded) instead of re-expanding and re-masking it —
/// the "recap of a recap = the same recap, updated" contract that stops recaps
/// from compounding across compactions.
pub(crate) const RECAP_HEADER_SENTINEL: &str = "via observation masking]";
const CLASSIFIED_RECAP_HEADER_SENTINEL: &str = "via classified compaction]";

/// Byte cap on a single carried-forward prior recap. Bounds the compound-growth
/// term independently of the per-message budget so a chain of compactions
/// converges instead of accreting.
const PRIOR_RECAP_CARRY_CAP: usize = 6_000;

/// Byte cap on the preview kept for an archived assistant turn. Assistant turns
/// carry the *calls* the model issued, not what it *learned*; keeping only a
/// short preview spends the recap budget on tool results (the observations)
/// rather than replaying verbatim call syntax.
const ASSISTANT_PREVIEW_CHARS: usize = 240;

/// Whether a message body is a previous observation-mask recap.
fn is_prior_recap(content: &str) -> bool {
    content.contains(RECAP_HEADER_SENTINEL) || content.contains(CLASSIFIED_RECAP_HEADER_SENTINEL)
}

/// Render one archived assistant turn as a bounded preview. Short turns pass
/// through; long turns keep a head slice plus a masked-marker tail so the drop
/// is visible in the same vocabulary as [`default_mask_tool_result`].
fn assistant_preview(content: &str) -> String {
    if content.len() <= ASSISTANT_PREVIEW_CHARS {
        return format!("[assistant] {content}");
    }
    let head = snap_to_line_end(content, ASSISTANT_PREVIEW_CHARS);
    let dropped = content.len().saturating_sub(head.len());
    format!("[assistant] {head}... [assistant turn truncated, {dropped} chars masked]")
}

/// Collapse runs of byte-identical consecutive lines into a single line with an
/// `(xN)` suffix. Turns repetitive call/read sequences ("looked at X 4 times")
/// into a count instead of N verbatim copies.
fn collapse_repeats(lines: Vec<String>) -> Vec<String> {
    let mut out: Vec<String> = Vec::with_capacity(lines.len());
    let mut run = 0usize;
    for line in lines {
        if out
            .last()
            .is_some_and(|prev| strip_repeat_suffix(prev) == line)
        {
            run += 1;
            let base =
                strip_repeat_suffix(out.last().expect("run implies a last line")).to_string();
            *out.last_mut().expect("run implies a last line") = format!("{base} (x{})", run + 1);
        } else {
            run = 0;
            out.push(line);
        }
    }
    out
}

#[expect(
    clippy::string_slice,
    reason = "idx is an rfind offset on the same line"
)]
fn strip_repeat_suffix(line: &str) -> &str {
    line.rfind(" (x")
        .filter(|_| line.ends_with(')'))
        .map(|idx| &line[..idx])
        .unwrap_or(line)
}

/// Deterministic observation-mask compaction.
#[cfg(test)]
pub(crate) fn observation_mask_compaction(
    old_messages: &[serde_json::Value],
    archived_count: usize,
) -> String {
    observation_mask_compaction_with_callback(
        old_messages,
        archived_count,
        None,
        DEFAULT_RECAP_BUDGET_BYTES,
    )
    .0
    .text
}

/// Test-only accessor exposing the recap body together with its
/// [`RecapMetrics`] and taking an explicit byte budget.
#[cfg(test)]
pub(crate) fn observation_mask_compaction_for_test(
    old_messages: &[serde_json::Value],
    archived_count: usize,
    budget_bytes: usize,
) -> (String, RecapMetrics) {
    let (summary, metrics) =
        observation_mask_compaction_with_callback(old_messages, archived_count, None, budget_bytes);
    (summary.text, metrics)
}

/// Build the observation-mask recap body under `budget_bytes`.
///
/// Contract (harn#4731):
///   - A single header line names how many messages were archived.
///   - The most-recent *prior* recap in the archive window is carried forward
///     once, bounded to [`PRIOR_RECAP_CARRY_CAP`], and never re-expanded, so
///     recaps do not compound across successive compactions.
///   - The remaining budget is spent NEWEST-first on load-bearing observations
///     (tool results — verify/test/build outcomes and the errors that preceded
///     edits, preserved by [`default_mask_tool_result`]) rather than on verbatim
///     assistant call replay, which is reduced to a short preview.
///   - Pinned `[no-compact]` grounding is always kept (already bounded to
///     [`MAX_PINNED_SEGMENTS`]).
///   - Overflow past the budget is dropped and summarized in one trailing
///     masked-marker line; repetitive identical lines collapse to `(xN)`.
fn observation_mask_compaction_with_callback(
    old_messages: &[serde_json::Value],
    archived_count: usize,
    mask_results: Option<&[Option<String>]>,
    budget_bytes: usize,
) -> (CompactionSummary, RecapMetrics) {
    let header =
        format!("[auto-compacted {archived_count} older messages via observation masking]");
    let pinned = latest_pinned_indices(old_messages.iter(), |msg| {
        msg.get("content").and_then(|v| v.as_str())
    });
    // Carry forward only the single most-recent prior recap; older recaps in the
    // window are already folded into it and compact as ordinary text.
    let prior_recap_idx = old_messages
        .iter()
        .enumerate()
        .rev()
        .find(|(_, msg)| {
            msg.get("content")
                .and_then(|v| v.as_str())
                .is_some_and(is_prior_recap)
        })
        .map(|(idx, _)| idx);

    let mut metrics = RecapMetrics {
        budget_bytes,
        ..RecapMetrics::default()
    };
    // Render newest-first so the budget lands on the most decision-relevant tail;
    // reverse to chronological order for output.
    let mut rendered_rev: Vec<String> = Vec::new();
    let mut used = header.len();

    for (idx, msg) in old_messages.iter().enumerate().rev() {
        let role = msg.get("role").and_then(|v| v.as_str()).unwrap_or("user");
        let content = msg
            .get("content")
            .and_then(|v| v.as_str())
            .unwrap_or_default();
        if content.is_empty() {
            continue;
        }

        if Some(idx) == prior_recap_idx {
            let carried = snap_to_line_end(content, PRIOR_RECAP_CARRY_CAP);
            rendered_rev.push(format!("[prior recap] {carried}"));
            metrics.carried_prior_recap = true;
            continue;
        }

        // Pinned grounding is unconditional (and already bounded).
        if pinned.contains(&idx) {
            rendered_rev.push(format!("[{role}] {content}"));
            continue;
        }

        let (line, is_result) = if role == "assistant" {
            (assistant_preview(content), false)
        } else if content_should_preserve(content) {
            (format!("[{role}] {content}"), true)
        } else if let Some(Some(custom)) = mask_results.and_then(|r| r.get(idx)) {
            (custom.clone(), true)
        } else {
            (default_mask_tool_result(role, content), true)
        };

        if used + line.len() + 1 > budget_bytes {
            metrics.dropped_count += 1;
            continue;
        }
        used += line.len() + 1;
        if is_result {
            metrics.kept_results_count += 1;
        }
        rendered_rev.push(line);
    }

    let mut body: Vec<String> = rendered_rev.into_iter().rev().collect();
    body = collapse_repeats(body);
    let carried: usize = body.iter().map(String::len).sum();
    let mut parts = vec![header];
    parts.append(&mut body);
    if metrics.dropped_count > 0 {
        parts.push(format!(
            "[{} older message(s) dropped to fit recap budget]",
            metrics.dropped_count
        ));
    }
    let summary = parts.join("\n");
    metrics.recap_bytes = summary.len();
    let carried = carried.min(summary.len());
    let scaffold_bytes = summary.len() - carried;
    (CompactionSummary::new(summary, scaffold_bytes), metrics)
}

/// Invoke the mask_callback to get per-message custom masks.
async fn invoke_mask_callback(
    ctx: Option<&AsyncBuiltinCtx>,
    callback: &VmValue,
    old_messages: &[serde_json::Value],
) -> Result<Vec<Option<String>>, VmError> {
    let VmValue::Closure(closure) = callback.clone() else {
        return Err(VmError::Runtime(
            "mask_callback must be a closure".to_string(),
        ));
    };
    let Some(ctx) = ctx else {
        return Err(VmError::Runtime(
            "mask_callback requires an async builtin VM context".to_string(),
        ));
    };
    let mut vm = ctx.child_vm();
    let messages_vm = VmValue::List(std::sync::Arc::new(
        old_messages
            .iter()
            .map(crate::stdlib::json_to_vm_value)
            .collect(),
    ));
    let result = vm.call_closure_pub(&closure, &[messages_vm]).await?;
    ctx.forward_output(&vm.take_output());
    let list = match result {
        VmValue::List(items) => items,
        _ => return Ok(vec![None; old_messages.len()]),
    };
    Ok(list
        .iter()
        .map(|v| match v {
            VmValue::String(s) => Some(s.to_string()),
            VmValue::Nil => None,
            _ => None,
        })
        .collect())
}

/// Rewrite each tool-result message in `messages` whose content exceeds
/// `config.tool_output_max_chars`, using `config.compress_callback` when set
/// (and a VM context is available) else the deterministic
/// [`microcompact_tool_output`]. Only the `content` text is replaced; the
/// message's `role`/`tool_call_id` are left untouched so tool-call pairing is
/// preserved. A `tool_output_max_chars` of 0 disables the pass.
async fn clamp_tool_outputs(
    ctx: Option<&AsyncBuiltinCtx>,
    messages: &mut [serde_json::Value],
    config: &AutoCompactConfig,
) -> Result<(), VmError> {
    if config.tool_output_max_chars == 0 {
        return Ok(());
    }
    // Exempt the most-recent pinned tool-outputs (those carrying the host's
    // `[no-compact]` marker) from length-clamping so the agent's live file view
    // stays intact. Bounded to the latest MAX_PINNED_SEGMENTS so older pinned
    // snapshots in the kept window still clamp and can't blow the budget.
    let pinned = latest_pinned_indices(messages.iter(), |msg| {
        if msg.get("role").and_then(|role| role.as_str()) == Some("tool") {
            msg.get("content").and_then(|content| content.as_str())
        } else {
            None
        }
    });
    for (idx, message) in messages.iter_mut().enumerate() {
        if message.get("role").and_then(|role| role.as_str()) != Some("tool") {
            continue;
        }
        let Some(content) = message.get("content").and_then(|content| content.as_str()) else {
            continue;
        };
        if content.len() <= config.tool_output_max_chars {
            continue;
        }
        if pinned.contains(&idx) {
            continue;
        }
        let content = content.to_string();
        let replacement = match (config.compress_callback.as_ref(), ctx) {
            (Some(callback), Some(ctx)) => {
                invoke_compress_callback(ctx, callback, &content, config.tool_output_max_chars)
                    .await?
            }
            _ => microcompact_tool_output(&content, config.tool_output_max_chars),
        };
        message["content"] = serde_json::Value::String(replacement);
    }
    Ok(())
}

/// Invoke `compress_callback(content, max_chars)` to replace one oversized
/// tool-output body, mirroring [`invoke_mask_callback`]'s child-VM closure
/// invocation. A non-string return falls back to the deterministic primitive.
async fn invoke_compress_callback(
    ctx: &AsyncBuiltinCtx,
    callback: &VmValue,
    content: &str,
    max_chars: usize,
) -> Result<String, VmError> {
    let VmValue::Closure(closure) = callback.clone() else {
        return Err(VmError::Runtime(
            "compress_callback must be a closure".to_string(),
        ));
    };
    let mut vm = ctx.child_vm();
    let args = [
        VmValue::String(arcstr::ArcStr::from(content)),
        VmValue::Int(max_chars as i64),
    ];
    let result = vm.call_closure_pub(&closure, &args).await?;
    ctx.forward_output(&vm.take_output());
    match result {
        VmValue::String(text) => Ok(text.to_string()),
        _ => Ok(microcompact_tool_output(content, max_chars)),
    }
}

#[derive(Clone, Copy)]
struct CompactionStrategyInputs<'a> {
    ctx: Option<&'a AsyncBuiltinCtx>,
    strategy: &'a CompactStrategy,
    old_messages: &'a [serde_json::Value],
    retained_messages: &'a [serde_json::Value],
    archived_count: usize,
    llm_opts: Option<&'a crate::llm::api::LlmCallOptions>,
    custom_compactor: Option<&'a VmValue>,
    custom_compactor_reminders: &'a [VmValue],
    mask_callback: Option<&'a VmValue>,
    summarize_prompt: Option<&'a str>,
    policy: &'a CompactionPolicy,
    recap_budget_bytes: usize,
}

/// Apply a single compaction strategy to a list of archived messages. Returns
/// the summary text plus [`RecapMetrics`] for the observation-mask strategy
/// (the only strategy that spends a recap budget); other strategies return
/// `None`.
async fn apply_compaction_strategy(
    input: CompactionStrategyInputs<'_>,
) -> Result<(CompactionSummary, Option<RecapMetrics>), VmError> {
    let CompactionStrategyInputs {
        strategy,
        old_messages,
        retained_messages,
        archived_count,
        llm_opts,
        custom_compactor,
        custom_compactor_reminders,
        mask_callback,
        summarize_prompt,
        policy,
        recap_budget_bytes,
        ctx,
    } = input;
    match strategy {
        CompactStrategy::Truncate => Ok((
            truncate_compaction_summary(old_messages, archived_count),
            None,
        )),
        CompactStrategy::Llm => llm_compaction_summary(
            old_messages,
            retained_messages,
            archived_count,
            llm_opts.ok_or_else(|| {
                VmError::Runtime(
                    "LLM transcript compaction requires active LLM call options".to_string(),
                )
            })?,
            summarize_prompt,
            policy,
        )
        .await
        .map(|summary| (summary, None)),
        CompactStrategy::Custom => custom_compaction_summary(
            ctx,
            old_messages,
            archived_count,
            custom_compactor.ok_or_else(|| {
                VmError::Runtime(
                    "compact_callback is required when compact_strategy is 'custom'".to_string(),
                )
            })?,
            custom_compactor_reminders,
            policy,
        )
        .await
        .map(|summary| (summary, None)),
        CompactStrategy::ObservationMask => {
            let mask_results = if let Some(cb) = mask_callback {
                Some(invoke_mask_callback(ctx, cb, old_messages).await?)
            } else {
                None
            };
            let (summary, metrics) = observation_mask_compaction_with_callback(
                old_messages,
                archived_count,
                mask_results.as_deref(),
                recap_budget_bytes,
            );
            Ok((summary, Some(metrics)))
        }
        CompactStrategy::Classify => Err(VmError::Runtime(
            "classify is a primary compaction strategy, not a positional fallback".into(),
        )),
    }
}

async fn apply_compaction_strategy_with_fallback(
    input: CompactionStrategyInputs<'_>,
    fallback_strategy: Option<&CompactStrategy>,
) -> Result<(CompactionSummary, CompactStrategy, Option<RecapMetrics>), VmError> {
    match apply_compaction_strategy(input).await {
        Ok((summary, metrics)) => Ok((summary, input.strategy.clone(), metrics)),
        Err(primary_error) => {
            if primary_error.is_uncatchable_control_flow()
                || crate::value::error_to_category(&primary_error)
                    == crate::value::ErrorCategory::Cancelled
            {
                return Err(primary_error);
            }
            let Some(fallback) = fallback_strategy.filter(|fallback| *fallback != input.strategy)
            else {
                return Err(primary_error);
            };
            let fallback_input = CompactionStrategyInputs {
                strategy: fallback,
                ..input
            };
            apply_compaction_strategy(fallback_input)
                .await
                .map(|(summary, metrics)| (summary, fallback.clone(), metrics))
        }
    }
}

#[derive(Debug)]
pub(crate) struct AutoCompactResult {
    pub summary: String,
    pub strategy: CompactStrategy,
    pub recap_metrics: Option<RecapMetrics>,
    pub classification: Option<Box<ClassificationReceipt>>,
    /// Typed source-window/summary measurement for this compaction. Always
    /// populated on this path, so a `None` field downstream means the
    /// measurement did not travel, never that it read zero.
    pub measurement: CompactionSourceMeasurement,
}

/// Auto-compact a message list in place using two-tier compaction.
#[cfg(test)]
pub(crate) async fn auto_compact_messages_with_result(
    messages: &mut Vec<serde_json::Value>,
    config: &AutoCompactConfig,
    llm_opts: Option<&crate::llm::api::LlmCallOptions>,
) -> Result<Option<AutoCompactResult>, VmError> {
    let Some((result, candidate)) =
        prepare_compaction_messages_with_ctx(None, messages, config, llm_opts).await?
    else {
        return Ok(None);
    };
    *messages = candidate;
    Ok(Some(result))
}

pub(crate) async fn prepare_compaction_messages_with_ctx(
    ctx: Option<&AsyncBuiltinCtx>,
    messages: &[serde_json::Value],
    config: &AutoCompactConfig,
    llm_opts: Option<&crate::llm::api::LlmCallOptions>,
) -> Result<Option<(AutoCompactResult, Vec<serde_json::Value>)>, VmError> {
    if config.compact_strategy == CompactStrategy::Classify && config.classification.is_none() {
        return Err(VmError::Runtime(
            "classify requires a classification policy".into(),
        ));
    }
    if config.hard_limit_strategy == CompactStrategy::Classify
        || config.fallback_strategy == Some(CompactStrategy::Classify)
    {
        return Err(VmError::Runtime(
            "classify cannot be a hard-limit or fallback strategy".into(),
        ));
    }
    if config.token_threshold > 0 && estimate_message_tokens(messages) <= config.token_threshold {
        return Ok(None);
    }
    if messages.len() <= config.keep_first.saturating_add(config.keep_last) {
        return Ok(None);
    }
    let compact_start = config.keep_first.min(messages.len());
    let original_split = messages.len().saturating_sub(config.keep_last);
    let mut split_at = original_split;
    // Snap back to a user-role boundary so the kept suffix begins at a clean
    // turn. OpenAI-compatible APIs reject tool results orphaned from their
    // assistant request, so splitting mid-turn corrupts the transcript.
    while split_at > compact_start
        && split_at < messages.len()
        && messages[split_at]
            .get("role")
            .and_then(|r| r.as_str())
            .is_none_or(|r| r != "user")
    {
        split_at -= 1;
    }
    // Fall back to the naive split (e.g. tool-heavy transcripts with the sole
    // user message at index 0) rather than skipping compaction entirely.
    if split_at == compact_start {
        split_at = original_split;
    }
    if let Some(volatile_start) = messages[split_at..]
        .iter()
        .position(is_reasoning_or_tool_turn_message)
        .map(|offset| split_at + offset)
    {
        if let Some(boundary) = volatile_start
            .checked_sub(1)
            .and_then(|idx| find_prev_user_boundary(messages, idx))
            .filter(|boundary| *boundary > compact_start)
        {
            split_at = boundary;
        }
    }
    // The naive fallback (and, in tool-heavy transcripts with no interior
    // user boundary, the volatile-start correction too) can still leave the
    // split pointing at a tool_result whose tool_use request would be
    // drained. Final pass: snap off any request/result pair.
    split_at = snap_split_off_tool_results(messages, split_at, compact_start);
    if split_at <= compact_start {
        return Ok(None);
    }
    // Mutating the caller before an awaited strategy succeeds loses source
    // context on refusal, cancellation, or a dropped future. Allocate only
    // after compaction is admitted, then publish the complete candidate once.
    let mut candidate = messages.to_vec();
    let result = Box::pin(compact_selected_window(
        ctx,
        &mut candidate,
        config,
        llm_opts,
        compact_start,
        split_at,
    ))
    .await?;
    Ok(Some((result, candidate)))
}

async fn compact_selected_window(
    ctx: Option<&AsyncBuiltinCtx>,
    messages: &mut Vec<serde_json::Value>,
    config: &AutoCompactConfig,
    llm_opts: Option<&crate::llm::api::LlmCallOptions>,
    compact_start: usize,
    split_at: usize,
) -> Result<AutoCompactResult, VmError> {
    let classification_anchor = if config.compact_strategy == CompactStrategy::Classify {
        classification::latest_user_anchor(messages)
    } else {
        String::new()
    };
    let old_messages: Vec<_> = messages.drain(compact_start..split_at).collect();
    let archived_count = old_messages.len();

    // Clamp oversized tool-result bodies in the *kept* window so the live
    // context honors the policy's `tool_output_max_chars` (and the
    // `compress_callback` override), not just the archived/summarized window.
    // Runs before the hard-limit estimate so tier-2 escalation
    // keys off the post-clamp size. Only the text body is rewritten; `role`
    // and `tool_call_id` are preserved so tool_call/tool_result pairing stays
    // intact.
    clamp_tool_outputs(ctx, messages, config).await?;

    let mut classification = None;
    let (mut summary, mut strategy, mut recap_metrics) =
        if config.compact_strategy == CompactStrategy::Classify {
            let classified = Box::pin(classification::classify_window(
                classification::ClassificationInputs {
                    ctx: ctx.ok_or_else(|| {
                        VmError::Runtime("classify requires an async VM context".into())
                    })?,
                    config: config.classification.as_deref().ok_or_else(|| {
                        VmError::Runtime("classify requires a classification policy".into())
                    })?,
                    archived: &old_messages,
                    retained: messages,
                    anchor: &classification_anchor,
                    first_index: compact_start,
                    budget_bytes: config.recap_budget_bytes,
                    active: llm_opts,
                    summarize_prompt: config.summarize_prompt.as_deref(),
                    policy: &config.policy,
                },
            ))
            .await?;
            let strategy = if classified.receipt.status == ClassificationStatus::Fallback {
                CompactStrategy::ObservationMask
            } else {
                CompactStrategy::Classify
            };
            classification = Some(classified.receipt);
            (classified.summary, strategy, classified.fallback_recap)
        } else {
            apply_compaction_strategy_with_fallback(
                CompactionStrategyInputs {
                    ctx,
                    strategy: &config.compact_strategy,
                    old_messages: &old_messages,
                    retained_messages: messages.as_slice(),
                    archived_count,
                    llm_opts,
                    custom_compactor: config.custom_compactor.as_ref(),
                    custom_compactor_reminders: &config.custom_compactor_reminders,
                    mask_callback: config.mask_callback.as_ref(),
                    summarize_prompt: config.summarize_prompt.as_deref(),
                    policy: &config.policy,
                    recap_budget_bytes: config.recap_budget_bytes,
                },
                config.fallback_strategy.as_ref(),
            )
            .await?
        };

    // A positional hard-limit pass may not undo keep/reword decisions or the
    // confidence floor. The classification receipt reports an unmet budget.
    if let Some(hard_limit) = config
        .hard_limit_tokens
        .filter(|_| classification.is_none())
    {
        let summary_msg = serde_json::json!({"role": "user", "content": &summary.text});
        let mut estimate_msgs = vec![summary_msg];
        estimate_msgs.extend_from_slice(messages.as_slice());
        let estimated = estimate_message_tokens(&estimate_msgs);
        if estimated > hard_limit {
            let tier1_as_messages = vec![serde_json::json!({
                "role": "user",
                "content": summary.text,
            })];
            let (hard_limit_summary, hard_limit_strategy, hard_limit_metrics) =
                apply_compaction_strategy_with_fallback(
                    CompactionStrategyInputs {
                        ctx,
                        strategy: &config.hard_limit_strategy,
                        old_messages: &tier1_as_messages,
                        retained_messages: messages.as_slice(),
                        archived_count,
                        llm_opts,
                        custom_compactor: config.custom_compactor.as_ref(),
                        custom_compactor_reminders: &config.custom_compactor_reminders,
                        mask_callback: None,
                        summarize_prompt: config.summarize_prompt.as_deref(),
                        policy: &config.policy,
                        recap_budget_bytes: config.recap_budget_bytes,
                    },
                    config.fallback_strategy.as_ref(),
                )
                .await?;
            summary = hard_limit_summary;
            strategy = hard_limit_strategy;
            // Tier-2 re-summarized the tier-1 recap; its metrics (if any)
            // describe the delivered body, so they supersede tier-1's.
            recap_metrics = hard_limit_metrics.or(recap_metrics);
        }
    }

    // Measure the source window against what the summary actually carried,
    // BEFORE the model-visible directives and the repair ledger are appended:
    // both are generated scaffold, and counting them would let a summary that
    // preserved nothing look substantial.
    let source_bytes: usize = old_messages.iter().map(message_content_bytes).sum();
    let measurement = CompactionSourceMeasurement {
        source_message_count: Some(archived_count),
        source_bytes: Some(source_bytes),
        summary_bytes: Some(summary.text.len()),
        carried_source_bytes: Some(summary.carried_source_bytes()),
    };

    // A non-empty source window whose summary carried zero source bytes is a
    // failed compaction, not a small one. Every summarizer has already taken its
    // one bounded deterministic retry by this point, so this is the terminal
    // path: refuse the candidate without publishing it, rather than replacing
    // real context with a bare header.
    if measurement.discarded_source_context() {
        return Err(VmError::Runtime(format!(
            "transcript compaction refused: the summary carried 0 of {source_bytes} source \
             bytes across {archived_count} archived messages (summary was {} bytes, all \
             generated scaffold); the source context was preserved",
            summary.text.len()
        )));
    }

    let summary = super::repair_ledger::append_repair_ledger_to_summary(
        apply_model_visible_policy(summary.text, &config.policy),
        &old_messages,
    );

    messages.insert(
        compact_start,
        serde_json::json!({
            "role": "user",
            "content": summary,
        }),
    );
    Ok(AutoCompactResult {
        summary,
        strategy,
        recap_metrics,
        classification,
        measurement,
    })
}

/// Auto-compact a message list in place using two-tier compaction.
#[cfg(test)]
pub(crate) async fn auto_compact_messages(
    messages: &mut Vec<serde_json::Value>,
    config: &AutoCompactConfig,
    llm_opts: Option<&crate::llm::api::LlmCallOptions>,
) -> Result<Option<String>, VmError> {
    Ok(
        auto_compact_messages_with_result(messages, config, llm_opts)
            .await?
            .map(|result| result.summary),
    )
}

fn apply_model_visible_policy(mut summary: String, policy: &CompactionPolicy) -> String {
    if !policy.is_model_visible_scope() {
        return summary;
    }
    let Some(directives) = policy.prompt_directives() else {
        return summary;
    };
    summary.push_str("\n\n[compaction instructions]\n");
    summary.push_str(&directives);
    summary
}

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