From bc9dac29f9b82a69e56418970c65437d6824d539 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Mon, 22 Jun 2026 13:26:26 +0530 Subject: [PATCH] feat(agent): native tool-output compaction (Stage 1a) with reversible CCR recovery (#3869) Co-authored-by: Claude --- .../agent/harness/compaction/cache_align.rs | 200 ++++++++++ .../agent/harness/compaction/demo.rs | 179 +++++++++ .../agent/harness/compaction/detect.rs | 301 +++++++++++++++ .../agent/harness/compaction/diff.rs | 218 +++++++++++ .../agent/harness/compaction/json_crusher.rs | 264 +++++++++++++ .../agent/harness/compaction/logs.rs | 294 +++++++++++++++ .../agent/harness/compaction/measure.rs | 150 ++++++++ src/openhuman/agent/harness/compaction/mod.rs | 350 ++++++++++++++++++ .../agent/harness/compaction/signals.rs | 141 +++++++ .../agent/harness/compaction/store.rs | 135 +++++++ src/openhuman/agent/harness/mod.rs | 1 + .../agent/harness/session/agent_tool_exec.rs | 13 + .../harness/session/builder/builder_tests.rs | 34 +- .../agent/harness/session/builder/factory.rs | 10 + .../agent/harness/session/builder/mod.rs | 13 + .../agent/harness/session/turn/core.rs | 1 + .../agent/harness/session/turn/tools.rs | 1 + .../harness/session/turn_engine_adapter.rs | 4 + .../harness/subagent_runner/tool_prep.rs | 75 ++++ src/openhuman/config/schema/context.rs | 16 + .../config/schema/load/env_overlay.rs | 13 + src/openhuman/config/schema/load_tests.rs | 24 ++ src/openhuman/context/manager.rs | 31 +- src/openhuman/tools/impl/system/mod.rs | 2 + .../tools/impl/system/retrieve_tool_output.rs | 115 ++++++ src/openhuman/tools/ops.rs | 4 + 26 files changed, 2586 insertions(+), 3 deletions(-) create mode 100644 src/openhuman/agent/harness/compaction/cache_align.rs create mode 100644 src/openhuman/agent/harness/compaction/demo.rs create mode 100644 src/openhuman/agent/harness/compaction/detect.rs create mode 100644 src/openhuman/agent/harness/compaction/diff.rs create mode 100644 src/openhuman/agent/harness/compaction/json_crusher.rs create mode 100644 src/openhuman/agent/harness/compaction/logs.rs create mode 100644 src/openhuman/agent/harness/compaction/measure.rs create mode 100644 src/openhuman/agent/harness/compaction/mod.rs create mode 100644 src/openhuman/agent/harness/compaction/signals.rs create mode 100644 src/openhuman/agent/harness/compaction/store.rs create mode 100644 src/openhuman/tools/impl/system/retrieve_tool_output.rs diff --git a/src/openhuman/agent/harness/compaction/cache_align.rs b/src/openhuman/agent/harness/compaction/cache_align.rs new file mode 100644 index 000000000..7cd3aa77c --- /dev/null +++ b/src/openhuman/agent/harness/compaction/cache_align.rs @@ -0,0 +1,200 @@ +//! System-prompt cache-alignment detector (warn-only). +//! +//! Clean-room port of headroom's `CacheAligner` (Apache-2.0) in its +//! **detector-only** form: it never mutates the prompt (that would itself bust +//! the cache prefix). It scans the cache-hot zone — the system prompt — for +//! *volatile* tokens that change every launch and therefore silently prevent +//! the provider KV-cache prefix from hitting: +//! +//! - UUIDs (canonical 36-char form) +//! - ISO-8601 timestamps +//! - JWTs (three base64url segments) +//! - hex hashes (MD5/SHA1/SHA256 lengths) +//! +//! When any are found it emits one warning log line so the volatility is +//! visible. OpenHuman already takes care to keep the system prompt stable for +//! KV-cache reuse (see `with_openhuman_thread_id` and the delegation-refresh +//! "system prompt unchanged for KV cache" path); this is the diagnostic that +//! catches regressions where dynamic content leaks back into the prefix. + +/// One detected volatile token: its kind and a short redacted sample. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct VolatileFinding { + pub kind: &'static str, + pub sample: String, +} + +/// Scan a system prompt for volatile tokens. Returns the findings (empty when +/// the prefix is stable). Pure — callers decide whether/how to log. +pub fn detect_volatile(system_prompt: &str) -> Vec { + let mut findings = Vec::new(); + // Delimiter = anything that can't appear inside the tokens we detect. + // Allowed inner chars: alphanumerics plus `- . : _` (UUID dashes, ISO + // timestamp `-`/`:`, JWT `.`/`-`/`_`). So `session=` splits cleanly. + for tok in system_prompt + .split(|c: char| !(c.is_ascii_alphanumeric() || matches!(c, '-' | '.' | ':' | '_'))) + { + if tok.len() < 8 { + continue; + } + if is_uuid(tok) { + findings.push(VolatileFinding { + kind: "uuid", + sample: redact(tok), + }); + } else if is_jwt(tok) { + findings.push(VolatileFinding { + kind: "jwt", + sample: redact(tok), + }); + } else if is_iso8601(tok) { + findings.push(VolatileFinding { + kind: "iso8601", + sample: tok.to_string(), + }); + } else if is_hex_hash(tok) { + findings.push(VolatileFinding { + kind: "hex_hash", + sample: redact(tok), + }); + } + } + findings +} + +/// Detect volatile tokens and, if any, emit a single warning log line. +/// Returns the number of findings (0 when the prefix is clean). +pub fn warn_if_volatile(system_prompt: &str) -> usize { + let findings = detect_volatile(system_prompt); + if !findings.is_empty() { + let mut kinds: Vec<&str> = findings.iter().map(|f| f.kind).collect(); + kinds.sort_unstable(); + kinds.dedup(); + ::log::warn!( + "[compaction][cache-align] system prompt contains {} volatile token(s) ({}) — KV-cache prefix may not hit; keep dynamic content out of the system prompt", + findings.len(), + kinds.join(", "), + ); + } + findings.len() +} + +fn redact(tok: &str) -> String { + let head: String = tok.chars().take(4).collect(); + format!("{head}…") +} + +/// Canonical RFC-4122 UUID: 8-4-4-4-12 hex with dashes (36 chars). The dashless +/// 32-char form is deliberately *not* accepted — it's structurally identical to +/// an MD5 digest and would mis-classify. +fn is_uuid(tok: &str) -> bool { + if tok.len() != 36 { + return false; + } + let bytes = tok.as_bytes(); + for (i, b) in bytes.iter().enumerate() { + let expect_dash = matches!(i, 8 | 13 | 18 | 23); + if expect_dash { + if *b != b'-' { + return false; + } + } else if !b.is_ascii_hexdigit() { + return false; + } + } + true +} + +/// JWT shape: three base64url segments joined by `.`, each non-trivial. We do +/// not verify the signature — only the structure. +fn is_jwt(tok: &str) -> bool { + let segs: Vec<&str> = tok.split('.').collect(); + if segs.len() != 3 { + return false; + } + segs.iter().all(|s| { + s.len() >= 4 + && s.bytes() + .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_') + }) && tok.starts_with("ey") +} + +/// Hex hash: all hex digits, length 32 (MD5) / 40 (SHA1) / 64 (SHA256). +fn is_hex_hash(tok: &str) -> bool { + matches!(tok.len(), 32 | 40 | 64) && tok.bytes().all(|b| b.is_ascii_hexdigit()) +} + +/// ISO-8601-ish timestamp: `YYYY-MM-DDThh:mm:ss` (or a space separator). Every +/// numeric position is validated so junk like `2026-aa-bbTcc:dd:ee` is rejected. +fn is_iso8601(tok: &str) -> bool { + let b = tok.as_bytes(); + if tok.len() < 19 { + return false; + } + let digit = |i: usize| b[i].is_ascii_digit(); + digit(0) + && digit(1) + && digit(2) + && digit(3) + && b[4] == b'-' + && digit(5) + && digit(6) + && b[7] == b'-' + && digit(8) + && digit(9) + && (b[10] == b'T' || b[10] == b' ') + && digit(11) + && digit(12) + && b[13] == b':' + && digit(14) + && digit(15) + && b[16] == b':' + && digit(17) + && digit(18) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn detects_uuid_and_timestamp() { + let prompt = "You are an agent. session=550e8400-e29b-41d4-a716-446655440000 started 2026-06-19T15:08:00"; + let f = detect_volatile(prompt); + assert!(f.iter().any(|x| x.kind == "uuid"), "{f:?}"); + assert!(f.iter().any(|x| x.kind == "iso8601"), "{f:?}"); + } + + #[test] + fn detects_hash_and_jwt() { + let prompt = "commit d41d8cd98f00b204e9800998ecf8427e token eyJhbGc.eyJzdWIi.SflKxwRJ here"; + let f = detect_volatile(prompt); + assert!(f.iter().any(|x| x.kind == "hex_hash"), "{f:?}"); + assert!(f.iter().any(|x| x.kind == "jwt"), "{f:?}"); + } + + #[test] + fn iso8601_rejects_non_numeric_lookalikes() { + // Shape matches but the fields aren't digits — must not be flagged. + let f = detect_volatile("ref 2026-aa-bbTcc:dd:ee here"); + assert!(!f.iter().any(|x| x.kind == "iso8601"), "{f:?}"); + // A real timestamp still flags. + let g = detect_volatile("at 2026-06-21T12:34:56 today"); + assert!(g.iter().any(|x| x.kind == "iso8601"), "{g:?}"); + } + + #[test] + fn stable_prompt_is_clean() { + let prompt = "You are a helpful assistant. Be concise. Use the tools provided."; + assert!(detect_volatile(prompt).is_empty()); + assert_eq!(warn_if_volatile(prompt), 0); + } + + #[test] + fn dashless_md5_not_uuid() { + // 32-char hex must classify as hex_hash, never uuid. + let f = detect_volatile("x d41d8cd98f00b204e9800998ecf8427e y"); + assert!(f.iter().any(|x| x.kind == "hex_hash")); + assert!(!f.iter().any(|x| x.kind == "uuid")); + } +} diff --git a/src/openhuman/agent/harness/compaction/demo.rs b/src/openhuman/agent/harness/compaction/demo.rs new file mode 100644 index 000000000..b00b9098a --- /dev/null +++ b/src/openhuman/agent/harness/compaction/demo.rs @@ -0,0 +1,179 @@ +//! Human-eyeball demo of compaction on dummy data. +//! +//! Unlike [`super::measure`] (which asserts token deltas), this prints the +//! actual BEFORE → AFTER text for each content type so you can see exactly +//! what the model would receive. Run it with output: +//! +//! ```text +//! cargo test -p openhuman --lib compaction::demo -- --nocapture +//! ``` + +#![cfg(test)] + +use super::super::token_budget::estimate_tokens; +use super::{compact_tool_output, store}; +use std::fmt::Write as _; + +/// Print a labelled before/after block. Input is shown head+tail so the +/// terminal stays readable; the compacted output is shown in full. +fn show(title: &str, tool: &str, raw: &str) -> String { + let before = estimate_tokens(raw); + let out = compact_tool_output(raw.to_string(), tool, true); + let after = estimate_tokens(&out); + let pct = if before == 0 { + 0.0 + } else { + 100.0 * (1.0 - after as f64 / before as f64) + }; + + println!("\n══════════════════════════════════════════════════════════════"); + println!("▶ {title} (tool={tool})"); + println!( + " {} bytes / ~{} tok → {} bytes / ~{} tok ({:.0}% saved)", + raw.len(), + before, + out.len(), + after, + pct + ); + println!("─── INPUT (first 8 + last 2 lines) ──────────────────────────"); + print_head_tail(raw, 8, 2); + println!("─── COMPACTED (what the model sees) ─────────────────────────"); + println!("{out}"); + out +} + +fn print_head_tail(text: &str, head: usize, tail: usize) { + let lines: Vec<&str> = text.lines().collect(); + if lines.len() <= head + tail { + for l in &lines { + println!("{l}"); + } + return; + } + for l in &lines[..head] { + println!("{l}"); + } + println!(" … {} lines …", lines.len() - head - tail); + for l in &lines[lines.len() - tail..] { + println!("{l}"); + } +} + +#[test] +fn demo_all_content_types() { + // 1 ── grep / code search is INTENTIONALLY NOT compacted (completeness + // tool). Demonstrate that it passes through untouched even when large. + let mut grep = String::from("48 match(es); scanned 4 file(s)\n"); + for f in 0..4 { + for i in 1..=12 { + let _ = writeln!( + grep, + "src/feature_{f}/service.rs:{i}: let outcome = dispatch_request(&context, request_payload_{i})?;" + ); + } + } + let grep_out = compact_tool_output(grep.clone(), "grep", true); + println!("\n══════════════════════════════════════════════════════════════"); + println!("▶ Code search (grep) — intentionally NOT compacted"); + println!( + " {} bytes → {} bytes (unchanged: {})", + grep.len(), + grep_out.len(), + grep_out == grep + ); + + // 2 ── build/test log: noise + warnings + a real error + summary. + let mut log = String::new(); + for i in 0..120 { + let _ = writeln!(log, " Compiling some_dependency_{i} v1.{i}.0"); + } + for i in 0..30 { + let _ = writeln!( + log, + "warning: unused variable `scratch` (occurrence {i}) at src/util.rs" + ); + } + let _ = writeln!(log, "error[E0382]: borrow of moved value `session`"); + let _ = writeln!(log, " --> src/agent/loop.rs:88:21"); + let _ = writeln!(log, " |"); + let _ = writeln!( + log, + " = note: move occurs because `session` has type `Session`" + ); + let _ = writeln!(log, "error: aborting due to previous error"); + let _ = writeln!(log, "test result: FAILED. 142 passed; 1 failed; 3 ignored"); + show("Build/test log (run_tests)", "run_tests", &log); + + // 3 ── git diff: small change wrapped in lots of unchanged context. + // (Fixtures are sized above MIN_BYTES_TO_COMPRESS = 2048 so the + // compressors actually engage — smaller outputs pass through untouched.) + let mut diff = + String::from("diff --git a/src/router.rs b/src/router.rs\n@@ -10,84 +10,85 @@\n"); + for i in 0..40 { + let _ = writeln!( + diff, + " // unchanged surrounding context line {i} carried along by the diff" + ); + } + let _ = writeln!(diff, "- route.register(\"/old\", old_handler);"); + let _ = writeln!(diff, "+ route.register(\"/new\", new_handler);"); + let _ = writeln!(diff, "+ route.register(\"/extra\", extra_handler);"); + for i in 0..40 { + let _ = writeln!( + diff, + " // trailing unchanged context line {i} after the change" + ); + } + show("Git diff (read_diff)", "read_diff", &diff); + + // 4 ── JSON list, small enough to stay lossless (under the row-drop + // threshold) but big enough to clear the byte floor → lossless table. + let mut small = Vec::new(); + for i in 0..30 { + small.push(format!( + r#"{{"id":{i},"name":"widget_{i}","status":"in_stock","warehouse":"east-1","sku":"WH-{i:04}"}}"# + )); + } + show( + "JSON list — 30 rows (lossless table)", + "list_inventory", + &format!("[{}]", small.join(",")), + ); + + // 5 ── JSON list, large (row-drop + CCR retrieval marker). + let mut big = Vec::new(); + for i in 0..80 { + big.push(format!( + r#"{{"id":{i},"name":"account_{i}","email":"a{i}@example.com","tier":"gold"}}"# + )); + } + let big_input = format!("[{}]", big.join(",")); + let out = show( + "JSON list — 80 rows (row-drop + CCR)", + "list_accounts", + &big_input, + ); + + // ── Demonstrate the reversibility: pull the hash from the marker and + // retrieve the full original via the CCR store (what the + // `retrieve_tool_output` tool does). + if let Some(marker) = out.lines().find(|l| l.contains("retrieve_tool_output(")) { + let hash = marker + .split("retrieve_tool_output(\"") + .nth(1) + .and_then(|s| s.split('"').next()) + .unwrap_or(""); + println!("\n─── CCR RETRIEVE round-trip ─────────────────────────────────"); + println!(" marker hash = {hash}"); + match store::retrieve(hash) { + Some(original) => println!( + " retrieve_tool_output(\"{hash}\") → {} bytes restored (matches original: {})", + original.len(), + original == big_input + ), + None => println!(" (evicted)"), + } + } + println!("\n══════════════════════════════════════════════════════════════\n"); +} diff --git a/src/openhuman/agent/harness/compaction/detect.rs b/src/openhuman/agent/harness/compaction/detect.rs new file mode 100644 index 000000000..bb9e63882 --- /dev/null +++ b/src/openhuman/agent/harness/compaction/detect.rs @@ -0,0 +1,301 @@ +//! Content-type detection and tool-name routing for compaction. +//! +//! Clean-room port of the routing behavior in headroom's `content_detector` +//! (Apache-2.0): cheap structural heuristics that classify a tool-output blob +//! so [`super::compact_tool_output`] can pick the right compressor. The tool +//! name gives a strong prior ([`ContentHint`]); detection validates/overrides +//! it for the `Auto` case. + +/// The kind of content a tool produced, after detection. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ContentType { + /// grep / ripgrep style `path:line:content` matches. + Search, + /// Build / test / lint output (compiler logs, test runners). + Log, + /// Unified git diff / patch. + Diff, + /// JSON array of objects (handled by the Phase-2 crusher). + JsonArray, + /// Nothing we compress — pass through unchanged. + PlainText, +} + +/// A prior derived from the tool name at the call site, so we don't have to +/// detect from scratch for the common case. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ContentHint { + Search, + Log, + Diff, + Json, + Auto, +} + +/// Map a tool name to its content prior. Mirrors the target-tool table in the +/// compaction plan. Unknown tools fall through to [`ContentHint::Auto`]. +pub fn hint_for_tool(tool_name: &str) -> ContentHint { + match tool_name { + "grep" | "glob_search" => ContentHint::Search, + // Tools whose output is *reliably* a build/test/lint log. + "run_tests" | "run_linter" | "npm_exec" | "node_exec" | "install_tool" | "lsp" => { + ContentHint::Log + } + "read_diff" | "git_operations" => ContentHint::Diff, + // `shell` is generic — its output is often NOT a log (find, seq, cat, + // generated CSV, a script printing a list). Route it through detection + // so only output that actually looks like a log gets log-compressed; + // anything else passes through. See the log compressor's no-signal guard. + _ => ContentHint::Auto, + } +} + +/// Resolve the content type to compress as. A non-`Auto` hint is trusted +/// unless detection strongly disagrees (e.g. a `shell` call that printed a +/// diff). For `Auto`, run full detection. +pub fn resolve(hint: ContentHint, content: &str) -> ContentType { + match hint { + ContentHint::Search => { + // Trust the hint, but if the body is clearly a diff/log prefer that. + if looks_like_diff(content) { + ContentType::Diff + } else { + ContentType::Search + } + } + ContentHint::Log => { + if looks_like_diff(content) { + ContentType::Diff + } else if search_line_ratio(content) >= 0.6 { + ContentType::Search + } else { + ContentType::Log + } + } + ContentHint::Diff => { + if looks_like_diff(content) { + ContentType::Diff + } else { + detect(content) + } + } + ContentHint::Json => { + if looks_like_json_array(content) { + ContentType::JsonArray + } else { + detect(content) + } + } + ContentHint::Auto => detect(content), + } +} + +/// Full structural detection, in priority order: JSON → diff → search → log → +/// plain text. Thresholds mirror headroom's detector. +pub fn detect(content: &str) -> ContentType { + let trimmed = content.trim_start(); + if trimmed.is_empty() { + return ContentType::PlainText; + } + if looks_like_json_array(content) { + return ContentType::JsonArray; + } + if looks_like_diff(content) { + return ContentType::Diff; + } + if search_line_ratio(content) >= 0.6 { + return ContentType::Search; + } + if log_line_ratio(content) >= 0.5 { + return ContentType::Log; + } + ContentType::PlainText +} + +/// True if `content` parses as a JSON array of objects (the crusher's input). +pub fn looks_like_json_array(content: &str) -> bool { + let trimmed = content.trim_start(); + if !trimmed.starts_with('[') { + return false; + } + match serde_json::from_str::(trimmed) { + Ok(serde_json::Value::Array(items)) => { + items.len() >= 2 && items.iter().any(|v| v.is_object()) + } + _ => false, + } +} + +/// True if `content` looks like a unified diff: a `diff --git` header or at +/// least one hunk header (`@@ ... @@`). +pub fn looks_like_diff(content: &str) -> bool { + let mut hunks = 0usize; + for line in content.lines().take(400) { + if line.starts_with("diff --git ") || line.starts_with("Index: ") { + return true; + } + if line.starts_with("@@ ") && line[3..].contains("@@") { + hunks += 1; + if hunks >= 1 { + return true; + } + } + } + false +} + +/// Fraction of non-empty lines that look like `path:line:...` search hits. +/// Handles a leading Windows drive letter (`C:\...`) so those paths aren't +/// mistaken for the line-number separator. +fn search_line_ratio(content: &str) -> f32 { + let mut total = 0usize; + let mut hits = 0usize; + for line in content.lines().take(2000) { + if line.trim().is_empty() { + continue; + } + total += 1; + if parse_search_line(line).is_some() { + hits += 1; + } + } + if total == 0 { + 0.0 + } else { + hits as f32 / total as f32 + } +} + +/// Fraction of lines carrying an error/warning indicator — the log signal. +fn log_line_ratio(content: &str) -> f32 { + use super::signals::{severity, Severity}; + let mut total = 0usize; + let mut hits = 0usize; + for line in content.lines().take(2000) { + if line.trim().is_empty() { + continue; + } + total += 1; + if severity(line) != Severity::Other { + hits += 1; + } + } + if total == 0 { + 0.0 + } else { + hits as f32 / total as f32 + } +} + +/// Parse a single grep/ripgrep line into `(path, line_number, content)`. +/// +/// Anchors on the earliest `::` marker, skipping a leading Windows +/// drive prefix (`C:`), so paths may contain `:` (drive), `-`, and spaces. +/// Returns `None` for context lines (`rg` `-` separators) and non-matches. +pub fn parse_search_line(line: &str) -> Option<(&str, u64, &str)> { + // Skip an optional Windows drive prefix like `C:` before scanning for the + // line-number marker. + let scan_from = if line.len() >= 2 { + let bytes = line.as_bytes(); + if bytes[0].is_ascii_alphabetic() && bytes[1] == b':' { + 2 + } else { + 0 + } + } else { + 0 + }; + + let rest = &line[scan_from..]; + // Find the first ':' that is followed by digits and then another ':'. + let mut search_start = 0usize; + while let Some(rel) = rest[search_start..].find(':') { + let colon = search_start + rel; + let after = &rest[colon + 1..]; + let digits_len = after.chars().take_while(|c| c.is_ascii_digit()).count(); + if digits_len > 0 && after.as_bytes().get(digits_len) == Some(&b':') { + let path = &line[..scan_from + colon]; + let num: u64 = after[..digits_len].parse().ok()?; + let body = &after[digits_len + 1..]; + if path.is_empty() { + return None; + } + return Some((path, num, body)); + } + search_start = colon + 1; + } + None +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn hints_map_known_tools() { + assert_eq!(hint_for_tool("grep"), ContentHint::Search); + assert_eq!(hint_for_tool("run_tests"), ContentHint::Log); + assert_eq!(hint_for_tool("read_diff"), ContentHint::Diff); + assert_eq!(hint_for_tool("file_read"), ContentHint::Auto); + // `shell` is generic → Auto (detection decides), not forced to Log. + assert_eq!(hint_for_tool("shell"), ContentHint::Auto); + } + + #[test] + fn detect_search_results() { + let c = + "src/main.rs:42:fn process() {\nsrc/lib.rs:7:pub use foo;\nsrc/x.rs:99: let y = 1;"; + assert_eq!(detect(c), ContentType::Search); + } + + #[test] + fn parse_unix_and_windows_paths() { + assert_eq!( + parse_search_line("src/main.rs:42:fn process() {"), + Some(("src/main.rs", 42, "fn process() {")) + ); + assert_eq!( + parse_search_line(r"C:\Users\me\a.rs:10:let x = 1;"), + Some((r"C:\Users\me\a.rs", 10, "let x = 1;")) + ); + // dashes in filename must survive + assert_eq!( + parse_search_line("pre-commit-config.yaml:3:foo"), + Some(("pre-commit-config.yaml", 3, "foo")) + ); + assert_eq!(parse_search_line("just a sentence"), None); + } + + #[test] + fn detect_diff() { + let c = "diff --git a/x.rs b/x.rs\n@@ -1,3 +1,4 @@\n+added\n-removed"; + assert_eq!(detect(c), ContentType::Diff); + } + + #[test] + fn detect_log() { + let c = + "Compiling foo\nwarning: unused\nerror[E0382]: borrow of moved value\nerror: aborting"; + assert_eq!(detect(c), ContentType::Log); + } + + #[test] + fn detect_json_array() { + let c = r#"[{"id":1,"name":"a"},{"id":2,"name":"b"}]"#; + assert_eq!(detect(c), ContentType::JsonArray); + } + + #[test] + fn plain_text_passes_through() { + assert_eq!( + detect("just some prose about a topic"), + ContentType::PlainText + ); + } + + #[test] + fn log_hint_with_search_body_routes_search() { + let body = "a.rs:1:x\nb.rs:2:y\nc.rs:3:z\nd.rs:4:w"; + assert_eq!(resolve(ContentHint::Log, body), ContentType::Search); + } +} diff --git a/src/openhuman/agent/harness/compaction/diff.rs b/src/openhuman/agent/harness/compaction/diff.rs new file mode 100644 index 000000000..6ce902b1b --- /dev/null +++ b/src/openhuman/agent/harness/compaction/diff.rs @@ -0,0 +1,218 @@ +//! Unified-diff compressor. +//! +//! Clean-room port of headroom's `DiffCompressor` (Apache-2.0), in the +//! lossy-but-bounded style of [`super::search`] / [`super::logs`] (the caller +//! offloads the original to CCR for retrieval): +//! +//! - **Always keep** structural lines: `diff --git`, `index`, `---`/`+++` +//! file headers, and `@@` hunk headers. +//! - **Always keep** changed lines (`+`/`-`) — they are the signal. +//! - **Collapse** long runs of unchanged context (lines starting with a +//! space) down to a few anchor lines plus a `[... N context lines ...]` +//! marker, so the model still sees where a change sits without paying for +//! the whole untouched neighbourhood. +//! - **Summarize** high-volume / low-value files (lockfiles, minified +//! bundles): the hunk body collapses to a one-line `+A/-B` summary. +//! +//! Changed lines are never dropped, so the diff stays faithful to *what +//! changed* even when the surrounding context is trimmed. + +use super::Compacted; +use std::fmt::Write as _; + +/// Context lines kept on each side of a changed run before collapsing. +pub const CONTEXT_ANCHOR: usize = 3; +/// A run of unchanged context longer than this collapses to a marker. +pub const CONTEXT_COLLAPSE_THRESHOLD: usize = 8; + +/// Compress a unified diff. Returns `None` when there's nothing structural to +/// work with or compression wouldn't shrink it. Lossy when it fires (collapses +/// context / summarizes hunks); the caller offloads the original to CCR. +pub fn compress(content: &str) -> Option { + let lines: Vec<&str> = content.lines().collect(); + if lines.is_empty() { + return None; + } + + let mut out = String::with_capacity(content.len() / 2 + 64); + let mut i = 0usize; + let mut current_file_is_noisy = false; + let mut saw_hunk = false; + + while i < lines.len() { + let line = lines[i]; + + // File header — note whether this file is a lockfile/bundle we summarize. + if line.starts_with("diff --git ") { + current_file_is_noisy = is_noisy_path(line); + let _ = writeln!(out, "{line}"); + i += 1; + continue; + } + if is_structural(line) { + saw_hunk |= line.starts_with("@@"); + // For a noisy file, collapse the entire hunk body to a summary. + if current_file_is_noisy && line.starts_with("@@") { + let _ = writeln!(out, "{line}"); + i += 1; + let (added, removed, consumed) = summarize_hunk_body(&lines[i..]); + let _ = writeln!( + out, + "[... lockfile/bundle hunk: +{added}/-{removed} line(s) omitted ...]" + ); + i += consumed; + continue; + } + let _ = writeln!(out, "{line}"); + i += 1; + continue; + } + + // Context line — collapse long unchanged runs. + if is_context(line) { + let start = i; + while i < lines.len() && is_context(lines[i]) { + i += 1; + } + let run = &lines[start..i]; + if run.len() > CONTEXT_COLLAPSE_THRESHOLD { + for l in &run[..CONTEXT_ANCHOR] { + let _ = writeln!(out, "{l}"); + } + let omitted = run.len() - 2 * CONTEXT_ANCHOR; + let _ = writeln!(out, "[... {omitted} context line(s) omitted ...]"); + for l in &run[run.len() - CONTEXT_ANCHOR..] { + let _ = writeln!(out, "{l}"); + } + } else { + for l in run { + let _ = writeln!(out, "{l}"); + } + } + continue; + } + + // Changed line (+/-) or anything else — keep verbatim. + let _ = writeln!(out, "{line}"); + i += 1; + } + + if !saw_hunk { + return None; + } + if out.len() >= content.len() { + return None; + } + log::debug!( + "[compaction][diff] {} -> {} bytes ({} input lines)", + content.len(), + out.len(), + lines.len(), + ); + Some(Compacted::lossy(out.trim_end().to_string())) +} + +/// Structural diff lines that must always survive. +fn is_structural(line: &str) -> bool { + line.starts_with("@@") + || line.starts_with("--- ") + || line.starts_with("+++ ") + || line.starts_with("index ") + || line.starts_with("new file") + || line.starts_with("deleted file") + || line.starts_with("rename ") + || line.starts_with("similarity ") + || line.starts_with("Binary files") +} + +/// A unified-diff context line: leading space, not a `+`/`-` change and not a +/// `---`/`+++` header (those are structural and handled first). +fn is_context(line: &str) -> bool { + line.starts_with(' ') +} + +/// Count added/removed lines in a hunk body until the next hunk/file header, +/// returning how many lines were consumed. +fn summarize_hunk_body(lines: &[&str]) -> (usize, usize, usize) { + let mut added = 0usize; + let mut removed = 0usize; + let mut n = 0usize; + for &line in lines { + if line.starts_with("@@") || line.starts_with("diff --git ") { + break; + } + if line.starts_with('+') && !line.starts_with("+++") { + added += 1; + } else if line.starts_with('-') && !line.starts_with("---") { + removed += 1; + } + n += 1; + } + (added, removed, n) +} + +/// Lockfiles and generated bundles whose diff body is rarely worth reading in +/// full. Matched against the `diff --git a/ b/` header. +fn is_noisy_path(diff_git_line: &str) -> bool { + let l = diff_git_line.to_ascii_lowercase(); + const NOISY: &[&str] = &[ + "cargo.lock", + "package-lock.json", + "pnpm-lock.yaml", + "yarn.lock", + "composer.lock", + "poetry.lock", + "gemfile.lock", + ".min.js", + ".min.css", + ".map", + "go.sum", + ]; + NOISY.iter().any(|p| l.contains(p)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn keeps_changed_lines_collapses_context() { + let mut s = String::from("diff --git a/x.rs b/x.rs\n@@ -1,40 +1,41 @@\n"); + for i in 0..20 { + let _ = writeln!(s, " context line {i} unchanged here"); + } + let _ = writeln!(s, "-old changed line"); + let _ = writeln!(s, "+new changed line"); + for i in 0..20 { + let _ = writeln!(s, " more context {i} unchanged"); + } + let out = compress(&s).expect("compresses").text; + assert!(out.contains("-old changed line"), "{out}"); + assert!(out.contains("+new changed line"), "{out}"); + assert!(out.contains("context line(s) omitted"), "{out}"); + assert!(out.contains("@@ -1,40 +1,41 @@")); + assert!(out.len() < s.len()); + } + + #[test] + fn summarizes_lockfile_hunk() { + let mut s = String::from("diff --git a/Cargo.lock b/Cargo.lock\n@@ -1,60 +1,80 @@\n"); + for i in 0..40 { + let _ = writeln!(s, "+ new dep entry {i}"); + } + for i in 0..20 { + let _ = writeln!(s, "- old dep entry {i}"); + } + let out = compress(&s).expect("compresses").text; + assert!(out.contains("lockfile/bundle hunk"), "{out}"); + assert!(out.contains("Cargo.lock")); + // Individual dep lines are gone. + assert!(!out.contains("new dep entry 7"), "{out}"); + assert!(out.len() < s.len()); + } + + #[test] + fn non_diff_returns_none() { + assert!(compress("just some text\nno hunks here").is_none()); + } +} diff --git a/src/openhuman/agent/harness/compaction/json_crusher.rs b/src/openhuman/agent/harness/compaction/json_crusher.rs new file mode 100644 index 000000000..6ecdfd88f --- /dev/null +++ b/src/openhuman/agent/harness/compaction/json_crusher.rs @@ -0,0 +1,264 @@ +//! JSON-array crusher. +//! +//! Clean-room port of the *lossless* core of headroom's `SmartCrusher` +//! (Apache-2.0): an array of objects that repeat the same keys is the single +//! most common bloated tool output (API list responses, DB rows, search +//! manifests). Re-rendering it as a table emits each key **once** instead of +//! per row, dropping the repeated key names and JSON punctuation. +//! +//! Up to [`ROW_DROP_THRESHOLD`] rows every value is preserved (nested values +//! render as compact JSON in their cell); the array→table reformat changes only +//! layout, so it returns [`Compacted::reformatted`]. Above the threshold the +//! table is additionally **row-dropped** (head + tail kept) and returns +//! [`Compacted::lossy`]. Either way the caller (`compact_tool_output`) offloads +//! the full original to CCR behind a `retrieve_tool_output("")` footer, so +//! the agent can always fetch the exact original JSON back on demand. + +use super::Compacted; +use serde_json::Value; +use std::fmt::Write as _; + +/// Minimum rows before tabular rendering is worth the header overhead. +pub const MIN_ROWS: usize = 3; +/// Above this many rows the table is *also* row-dropped: head + tail rows are +/// kept and the full original is offloaded to CCR behind a retrieval marker. +pub const ROW_DROP_THRESHOLD: usize = 40; +/// Rows kept from the head when row-dropping. +pub const HEAD_ROWS: usize = 20; +/// Rows kept from the tail when row-dropping. +pub const TAIL_ROWS: usize = 10; + +/// Compress a JSON array-of-objects into a compact table. Returns `None` when +/// the content isn't a uniform-enough array of objects or wouldn't shrink. +/// +/// Lossless (only reformats) up to [`ROW_DROP_THRESHOLD`] rows; above it the +/// middle rows are dropped and the result is marked `lossy` so the caller +/// offloads the original to CCR behind the retrieval footer. +pub fn compress(content: &str) -> Option { + let value: Value = serde_json::from_str(content.trim()).ok()?; + let array = value.as_array()?; + if array.len() < MIN_ROWS { + return None; + } + // Every element must be an object for a clean table; mixed arrays bail. + if !array.iter().all(Value::is_object) { + return None; + } + + // Column order = first-seen key order across all rows (union, stable). + let mut columns: Vec = Vec::new(); + for item in array { + if let Some(obj) = item.as_object() { + for key in obj.keys() { + if !columns.iter().any(|c| c == key) { + columns.push(key.clone()); + } + } + } + } + if columns.len() < 2 { + return None; + } + + // Render every row's cells up front so we can choose full vs. row-dropped. + let mut rows: Vec = Vec::with_capacity(array.len()); + for item in array { + let obj = item.as_object()?; + let cells: Vec = columns + .iter() + .map(|col| match obj.get(col) { + // Distinguish a truly absent key (blank) from an explicit null + // (rendered as `null` by render_cell) so the view stays faithful. + None => String::new(), + Some(v) => render_cell(v), + }) + .collect(); + rows.push(cells.join(" | ")); + } + + let lossy = rows.len() > ROW_DROP_THRESHOLD; + let mut out = String::with_capacity(content.len()); + let _ = writeln!( + out, + "[json table: {} rows × {} cols · blank=absent key · exact original via retrieve footer]", + rows.len(), + columns.len() + ); + let _ = writeln!(out, "{}", columns.join(" | ")); + + if lossy { + // Keep head + tail; the caller offloads the full original to CCR and + // appends the retrieve_tool_output footer, so the dropped middle stays + // recoverable. The inline marker here just shows where rows were cut. + let dropped = rows.len() - HEAD_ROWS - TAIL_ROWS; + for row in rows.iter().take(HEAD_ROWS) { + let _ = writeln!(out, "{row}"); + } + let _ = writeln!(out, "[... {dropped} middle rows omitted ...]"); + for row in rows.iter().skip(rows.len() - TAIL_ROWS) { + let _ = writeln!(out, "{row}"); + } + } else { + for row in &rows { + let _ = writeln!(out, "{row}"); + } + } + + let out = out.trim_end().to_string(); + if out.len() >= content.len() { + return None; + } + log::debug!( + "[compaction][json] {} rows × {} cols, lossy={} ({} -> {} bytes)", + rows.len(), + columns.len(), + lossy, + content.len(), + out.len(), + ); + if lossy { + Some(Compacted::lossy(out)) + } else { + // All values preserved, but the array→table reformat changes layout + // (key order, quoting). Reported as `reformatted` so the caller still + // offloads the original — the agent can fetch exact JSON bytes back. + Some(Compacted::reformatted(out)) + } +} + +/// Render a single cell. Scalars print bare-ish (strings unquoted unless they +/// contain the column separator); nested values stay as compact JSON so the +/// table remains lossless. +fn render_cell(v: &Value) -> String { + match v { + Value::String(s) if !s.contains('|') && !s.contains('\n') => s.clone(), + Value::Bool(b) => b.to_string(), + Value::Number(n) => n.to_string(), + other => serde_json::to_string(other).unwrap_or_default(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn crushes_uniform_array() { + let mut rows = Vec::new(); + for i in 0..20 { + rows.push(format!( + r#"{{"id":{i},"name":"item number {i}","status":"active","owner":"team-alpha"}}"# + )); + } + let input = format!("[{}]", rows.join(",")); + let out = compress(&input).expect("compresses").text; + // Header + key names appear once, not 20× (column order is whatever + // serde_json yields — don't assume insertion order here). + assert_eq!(out.matches("status").count(), 1, "{out}"); + for col in ["id", "name", "status", "owner"] { + assert!(out.lines().nth(1).unwrap().contains(col), "missing {col}"); + } + // Data preserved. + assert!(out.contains("item number 7")); + assert!(out.contains("19")); + assert!(out.len() < input.len(), "expected shrink"); + } + + #[test] + fn preserves_nested_values_losslessly() { + // Enough rows that the table beats the input even with the header. + let mut rows = Vec::new(); + for i in 0..8 { + rows.push(format!( + r#"{{"id":{i},"tags":["alpha","beta"],"meta":{{"k":{i},"label":"row{i}"}}}}"# + )); + } + let input = format!("[{}]", rows.join(",")); + let out = compress(&input).expect("compresses").text; + assert!(out.contains(r#"["alpha","beta"]"#), "{out}"); + assert!(out.contains(r#""label":"row3""#), "{out}"); + } + + #[test] + fn handles_missing_keys() { + // Enough rows with longish values that dropping repeated keys shrinks. + let mut rows = Vec::new(); + for i in 0..12 { + rows.push(format!( + r#"{{"alpha":{i},"bravo":"value string {i}","charlie":"another value {i}"}}"# + )); + } + rows.push(r#"{"alpha":99}"#.to_string()); // missing bravo/charlie + let input = format!("[{}]", rows.join(",")); + let out = compress(&input).expect("compresses").text; + let header = out.lines().nth(1).unwrap(); + for col in ["alpha", "bravo", "charlie"] { + assert!(header.contains(col), "header missing {col}: {header}"); + } + assert!(out.len() < input.len()); + } + + #[test] + fn large_array_row_drops_and_is_marked_lossy() { + let mut rows = Vec::new(); + for i in 0..200 { + rows.push(format!( + r#"{{"id":{i},"name":"record number {i}","status":"active","note":"some detail {i}"}}"# + )); + } + let input = format!("[{}]", rows.join(",")); + let c = compress(&input).expect("compresses"); + + // Marked lossy → the caller (compact_tool_output) offloads to CCR and + // appends the retrieve footer. The CCR round-trip is covered at the + // mod.rs level (lossy_outputs_are_recoverable). + assert!(c.lossy, "row-dropped output must be lossy"); + assert!(c.text.contains("middle rows omitted"), "{}", c.text); + + // Head + tail rows survive; the middle is dropped. + assert!(c.text.contains("record number 0"), "{}", c.text); + assert!(c.text.contains("record number 199"), "{}", c.text); + assert!( + !c.text.contains("record number 100"), + "middle should be dropped" + ); + assert!(c.text.len() < input.len()); + } + + #[test] + fn distinguishes_explicit_null_from_absent_key() { + // explicit null → "null"; absent key → blank. (Faithfulness: the two + // must not be conflated.) + let mut rows = Vec::new(); + for i in 0..10 { + rows.push(format!( + r#"{{"id":{i},"note":null,"tag":"long enough value to ensure shrink {i}"}}"# + )); + } + let input = format!("[{}]", rows.join(",")); + let c = compress(&input).expect("compresses"); + // A row with explicit null renders the literal "null" (not blank). + assert!(c.text.contains("null"), "{}", c.text); + } + + #[test] + fn small_table_is_lossless() { + let mut rows = Vec::new(); + for i in 0..10 { + rows.push(format!( + r#"{{"id":{i},"name":"row {i} with a reasonably long value","kind":"sample"}}"# + )); + } + let input = format!("[{}]", rows.join(",")); + let c = compress(&input).expect("compresses"); + assert!(!c.lossy, "a full table drops no data"); + assert!(!c.text.contains("omitted")); + } + + #[test] + fn non_array_returns_none() { + assert!(compress(r#"{"a":1}"#).is_none()); + assert!(compress("[1,2,3]").is_none()); + assert!(compress(r#"[{"a":1}]"#).is_none()); // too few rows + } +} diff --git a/src/openhuman/agent/harness/compaction/logs.rs b/src/openhuman/agent/harness/compaction/logs.rs new file mode 100644 index 000000000..f8497c6c0 --- /dev/null +++ b/src/openhuman/agent/harness/compaction/logs.rs @@ -0,0 +1,294 @@ +//! Build/test/lint log compressor. +//! +//! Clean-room port of headroom's `LogCompressor` (Apache-2.0). Keeps the +//! lines an agent acts on and drops the ceremony: +//! +//! - **errors** — keep up to [`MAX_ERRORS`] (biased to the first and last, +//! which are usually the root cause and the abort line). +//! - **warnings** — keep up to [`MAX_WARNINGS`], de-duplicated (the "same +//! deprecation × 47" case collapses to one). +//! - **stack traces** — keep up to [`MAX_STACK_TRACES`] runs of indented +//! frames, each capped at [`STACK_TRACE_MAX_LINES`]. +//! - **summary lines** — always keep test/build summaries +//! (`test result: ...`, `N passed; M failed`, `error: aborting`). +//! - hard cap of [`MAX_TOTAL_LINES`] kept lines overall. +//! +//! Output preserves original line order and notes how many lines were +//! dropped. Lossy-but-bounded: first/last errors and the summary always +//! survive. + +use super::signals::{severity, Severity}; +use super::Compacted; +use std::collections::HashSet; +use std::fmt::Write as _; + +pub const MAX_ERRORS: usize = 10; +pub const MAX_WARNINGS: usize = 5; +pub const MAX_STACK_TRACES: usize = 3; +pub const STACK_TRACE_MAX_LINES: usize = 20; +pub const MAX_TOTAL_LINES: usize = 100; + +/// Compress a build/test/lint log. Returns `None` when nothing can be saved +/// (few lines, or no reduction possible) so the caller passes it through. +/// Lossy when it fires (drops lines); the caller offloads the original to CCR. +pub fn compress(content: &str) -> Option { + let lines: Vec<&str> = content.lines().collect(); + if lines.len() <= MAX_TOTAL_LINES { + // Short enough already — the byte budget (if any) will handle it. + return None; + } + + // Indices we decide to keep. BTreeSet keeps output in original order. + let mut keep: std::collections::BTreeSet = std::collections::BTreeSet::new(); + + // 1. Summary lines — always keep. + for (i, line) in lines.iter().enumerate() { + if is_summary_line(line) { + keep.insert(i); + } + } + + // 2. Errors — first MAX_ERRORS/2 and last MAX_ERRORS/2 by appearance. + let error_idx: Vec = lines + .iter() + .enumerate() + .filter(|(_, l)| severity(l) == Severity::Error) + .map(|(i, _)| i) + .collect(); + for &i in select_first_last(&error_idx, MAX_ERRORS).iter() { + keep.insert(i); + } + + // 3. Warnings — de-duplicated by normalized text, capped. + let mut seen_warn: HashSet = HashSet::new(); + let mut warn_kept = 0usize; + for (i, line) in lines.iter().enumerate() { + if warn_kept >= MAX_WARNINGS { + break; + } + if severity(line) == Severity::Warning { + let norm = normalize_for_dedupe(line); + if seen_warn.insert(norm) { + keep.insert(i); + warn_kept += 1; + } + } + } + + // 4. Stack traces — runs of indented / "at "/"#n " frames following an + // error, capped in count and per-trace length. + let mut traces_kept = 0usize; + let mut i = 0usize; + while i < lines.len() && traces_kept < MAX_STACK_TRACES { + if is_stack_frame(lines[i]) { + let start = i; + let mut taken = 0usize; + while i < lines.len() && is_stack_frame(lines[i]) { + if taken < STACK_TRACE_MAX_LINES { + keep.insert(i); + taken += 1; + } + i += 1; + } + if i > start { + traces_kept += 1; + } + } else { + i += 1; + } + } + + if keep.is_empty() { + // No errors, warnings, stack traces, or summary lines — this almost + // certainly isn't a log (e.g. generic `shell` output: a file listing, + // CSV, or a script printing data). Do NOT head/tail-truncate it, which + // would silently drop the middle of legitimate data. Pass it through to + // the byte budget instead. + return None; + } + + // Enforce the global line cap, keeping the earliest + latest kept lines so + // the root cause and the final summary both survive. + let kept_vec: Vec = keep.iter().copied().collect(); + let kept_vec = if kept_vec.len() > MAX_TOTAL_LINES { + select_first_last(&kept_vec, MAX_TOTAL_LINES) + } else { + kept_vec + }; + let kept_set: std::collections::BTreeSet = kept_vec.into_iter().collect(); + + // Render with gap markers for runs of dropped lines. + let mut out = String::with_capacity(content.len() / 2 + 64); + let mut prev: Option = None; + let mut total_dropped = 0usize; + for &i in &kept_set { + if let Some(p) = prev { + let gap = i - p - 1; + if gap > 0 { + total_dropped += gap; + let _ = writeln!(out, "[... {gap} line(s) omitted ...]"); + } + } else if i > 0 { + total_dropped += i; + let _ = writeln!(out, "[... {i} line(s) omitted ...]"); + } + let _ = writeln!(out, "{}", lines[i]); + prev = Some(i); + } + if let Some(p) = prev { + let tail = lines.len().saturating_sub(p + 1); + if tail > 0 { + total_dropped += tail; + let _ = writeln!(out, "[... {tail} line(s) omitted ...]"); + } + } + + if out.len() >= content.len() { + return None; + } + log::debug!( + "[compaction][logs] kept {} of {} line(s), dropped {}", + kept_set.len(), + lines.len(), + total_dropped, + ); + Some(Compacted::lossy(out.trim_end().to_string())) +} + +/// Choose at most `cap` indices, biased to the first and last by value. +/// `idx` must be ascending. Keeps `ceil(cap/2)` from the front and the rest +/// from the back, preserving order and avoiding duplicates. +fn select_first_last(idx: &[usize], cap: usize) -> Vec { + if idx.len() <= cap { + return idx.to_vec(); + } + if cap == 0 { + return Vec::new(); + } + let head = cap.div_ceil(2); + let tail = cap - head; + let mut out: std::collections::BTreeSet = std::collections::BTreeSet::new(); + for &i in idx.iter().take(head) { + out.insert(i); + } + for &i in idx.iter().rev().take(tail) { + out.insert(i); + } + out.into_iter().collect() +} + +/// Test/build summary lines worth always keeping. +fn is_summary_line(line: &str) -> bool { + let l = line.to_ascii_lowercase(); + let l = l.trim(); + l.starts_with("test result:") + || l.starts_with("error: aborting") + || l.contains(" passed") + || l.contains(" failed") + || l.contains("tests passed") + || l.contains("tests failed") + || l.contains("failures:") + || (l.contains("warning") && l.contains("generated")) + || l.starts_with("error: could not compile") + || l.starts_with("build failed") + || l.starts_with("build succeeded") + || (l.contains("npm") && l.contains("err")) +} + +/// A stack-trace frame: leading whitespace + `at `/`#n `/`File "...` etc. +fn is_stack_frame(line: &str) -> bool { + let trimmed = line.trim_start(); + if trimmed.is_empty() { + return false; + } + let indented = line.starts_with(' ') || line.starts_with('\t'); + indented + && (trimmed.starts_with("at ") + || trimmed.starts_with("File \"") + || (trimmed.starts_with('#') && trimmed[1..].starts_with(|c: char| c.is_ascii_digit()))) +} + +/// Normalize a warning line for de-duplication: lowercase and strip digits so +/// "warning at line 12" and "warning at line 88" collapse together. +fn normalize_for_dedupe(line: &str) -> String { + line.chars() + .filter(|c| !c.is_ascii_digit()) + .collect::() + .to_ascii_lowercase() + .split_whitespace() + .collect::>() + .join(" ") +} + +#[cfg(test)] +mod tests { + use super::*; + + fn noisy_log() -> String { + let mut s = String::new(); + for i in 0..200 { + let _ = writeln!(s, " Compiling crate_{i} v0.1.0"); + } + let _ = writeln!(s, "error[E0382]: borrow of moved value `x`"); + let _ = writeln!(s, " --> src/main.rs:10:5"); + let _ = writeln!(s, "error: aborting due to previous error"); + let _ = writeln!(s, "test result: FAILED. 3 passed; 1 failed"); + s + } + + #[test] + fn keeps_errors_and_summary_drops_noise() { + let input = noisy_log(); + let out = compress(&input).expect("compresses").text; + assert!(out.contains("error[E0382]"), "{out}"); + assert!(out.contains("error: aborting"), "{out}"); + assert!(out.contains("test result: FAILED"), "{out}"); + assert!(out.lines().count() <= MAX_TOTAL_LINES + 10); + assert!(out.len() < input.len()); + assert!(out.contains("omitted")); + } + + #[test] + fn dedupes_warnings() { + let mut s = String::new(); + for i in 0..150 { + let _ = writeln!(s, "warning: unused variable at line {i}"); + } + let _ = writeln!(s, "test result: ok. 1 passed; 0 failed"); + let out = compress(&s).expect("compresses").text; + let warns = out.matches("unused variable").count(); + assert!(warns <= MAX_WARNINGS, "kept {warns} warnings"); + } + + #[test] + fn non_log_data_passes_through_not_head_tail_truncated() { + // Generic shell-style output with no errors/warnings/summary: a long + // listing. Must pass through (None) rather than dropping the middle. + let mut s = String::new(); + for i in 0..400 { + let _ = writeln!(s, "/var/data/file_{i:04}.bin\t{i}\trwxr-xr-x"); + } + assert!(compress(&s).is_none(), "non-log data must not be truncated"); + } + + #[test] + fn short_log_passes_through() { + let s = "line1\nline2\nerror: boom\n"; + assert!(compress(s).is_none()); + } + + #[test] + fn keeps_stack_trace_capped() { + let mut s = String::new(); + let _ = writeln!(s, "panicked at 'boom'"); + for i in 0..50 { + let _ = writeln!(s, " at frame_{i} (src/x.rs:{i})"); + } + for i in 0..120 { + let _ = writeln!(s, "info: step {i}"); + } + let out = compress(&s).expect("compresses").text; + let frames = out.matches(" at frame_").count(); + assert!(frames <= STACK_TRACE_MAX_LINES, "{frames} frames kept"); + } +} diff --git a/src/openhuman/agent/harness/compaction/measure.rs b/src/openhuman/agent/harness/compaction/measure.rs new file mode 100644 index 000000000..9b81aa89b --- /dev/null +++ b/src/openhuman/agent/harness/compaction/measure.rs @@ -0,0 +1,150 @@ +//! Deterministic token-savings harness for compaction (the `[agent_cost]` +//! A/B, measured at the compaction boundary). +//! +//! A live A/B runs the agent against the backend with `OPENHUMAN_COMPACTION=0` +//! vs on and diffs the `[agent_cost] … tokens_in=…` lines. That needs LLM +//! credentials + a real workspace, so it's the operator's job (commands in +//! `compaction-plan.md`). What we *can* pin down reproducibly is the input-side +//! reduction those `tokens_in` deltas come from: run representative tool +//! outputs through [`super::compact_tool_output`] and measure the token delta +//! with the **same estimator the budget/cost path uses** +//! ([`super::super::token_budget::estimate_tokens`]). +//! +//! These tests double as a regression guard: if a future change regresses the +//! savings on a fixture, they fail. Run with output: +//! +//! ```text +//! cargo test -p openhuman --lib compaction::measure -- --nocapture +//! ``` + +#![cfg(test)] + +use super::super::token_budget::estimate_tokens; +use super::compact_tool_output; +use std::fmt::Write as _; + +/// A single A/B sample: token counts before/after compaction for one fixture. +struct Sample { + label: &'static str, + tool: &'static str, + tokens_before: usize, + tokens_after: usize, +} + +impl Sample { + fn run(label: &'static str, tool: &'static str, raw: String) -> Self { + let before = estimate_tokens(&raw); + let compacted = compact_tool_output(raw, tool, true); + let after = estimate_tokens(&compacted); + Sample { + label, + tool, + tokens_before: before, + tokens_after: after, + } + } + + fn saved_pct(&self) -> f64 { + if self.tokens_before == 0 { + 0.0 + } else { + 100.0 * (1.0 - self.tokens_after as f64 / self.tokens_before as f64) + } + } +} + +// ── Representative fixtures (the loud tool families from the plan) ────────── +// Note: grep/search is intentionally not compacted, so it is not measured here. + +fn cargo_test_log_fixture() -> String { + let mut s = String::new(); + for i in 0..300 { + let _ = writeln!(s, " Compiling dependency_crate_{i} v0.{i}.0"); + } + for i in 0..40 { + let _ = writeln!( + s, + "warning: unused variable `tmp` at src/x.rs (occurrence {i})" + ); + } + let _ = writeln!(s, "error[E0382]: borrow of moved value `config`"); + let _ = writeln!(s, " --> src/server/boot.rs:142:18"); + let _ = writeln!(s, "error: aborting due to previous error"); + let _ = writeln!(s, "test result: FAILED. 87 passed; 1 failed; 0 ignored"); + s +} + +fn json_list_fixture() -> String { + let mut rows = Vec::new(); + for i in 0..150 { + rows.push(format!( + r#"{{"id":{i},"name":"user_{i}","email":"user{i}@example.com","status":"active","role":"member"}}"# + )); + } + format!("[{}]", rows.join(",")) +} + +fn diff_fixture() -> String { + let mut s = String::from("diff --git a/src/big.rs b/src/big.rs\n@@ -1,80 +1,82 @@\n"); + for i in 0..40 { + let _ = writeln!(s, " unchanged context line {i} that the diff carries along"); + } + let _ = writeln!(s, "- let x = old_implementation();"); + let _ = writeln!(s, "+ let x = new_implementation(with_args);"); + for i in 0..40 { + let _ = writeln!(s, " more unchanged context line {i} after the change"); + } + s +} + +#[test] +fn ab_token_savings_report() { + let samples = vec![ + Sample::run("cargo test failure", "run_tests", cargo_test_log_fixture()), + Sample::run("JSON list (150 rows)", "list_records", json_list_fixture()), + Sample::run("git diff (large context)", "read_diff", diff_fixture()), + ]; + + let mut total_before = 0usize; + let mut total_after = 0usize; + println!("\n[compaction A/B] token_in savings at the compaction boundary (estimate_tokens):"); + println!( + " {:<26} {:>10} {:>10} {:>9}", + "workload", "before", "after", "saved" + ); + for s in &samples { + total_before += s.tokens_before; + total_after += s.tokens_after; + println!( + " {:<26} {:>10} {:>10} {:>8.0}% (tool={})", + s.label, + s.tokens_before, + s.tokens_after, + s.saved_pct(), + s.tool + ); + // Each loud-family fixture must save meaningfully — regression guard. + assert!( + s.saved_pct() >= 30.0, + "{} only saved {:.0}%", + s.label, + s.saved_pct() + ); + } + let overall = 100.0 * (1.0 - total_after as f64 / total_before as f64); + println!( + " {:<26} {:>10} {:>10} {:>8.0}%", + "TOTAL", total_before, total_after, overall + ); + assert!(overall >= 50.0, "overall savings only {overall:.0}%"); +} + +#[test] +fn disabled_yields_zero_savings() { + // Sanity: with the kill-switch off, the harness sees no reduction — this is + // the control arm of the A/B. (Uses a fixture that *would* compact when on.) + let raw = cargo_test_log_fixture(); + let before = estimate_tokens(&raw); + let after = estimate_tokens(&compact_tool_output(raw, "run_tests", false)); + assert_eq!(before, after); +} diff --git a/src/openhuman/agent/harness/compaction/mod.rs b/src/openhuman/agent/harness/compaction/mod.rs new file mode 100644 index 000000000..96c171633 --- /dev/null +++ b/src/openhuman/agent/harness/compaction/mod.rs @@ -0,0 +1,350 @@ +//! Native tool-output compaction (Stage 1a). +//! +//! Content-aware compression of large tool outputs, applied in +//! `Agent::execute_tool_call` **before** the byte-cap truncation in +//! [`crate::openhuman::context::tool_result_budget`] (Stage 1) and before the +//! result enters conversation history. Operates on fresh bytes that have not +//! been sent to the backend, so — like Stage 1 — it never mutates +//! previously-sent history and cannot bust the provider KV-cache prefix. +//! +//! This is a clean-room Rust port of the deterministic (non-ML) compressors +//! from headroom (, Apache-2.0): +//! content routing + grep/log/diff compaction. The ML text/image compressors +//! are intentionally out of scope (no Python, no ONNX, no model download). +//! +//! See `compaction-plan.md` for the full design. The downstream byte cap lives +//! in [`crate::openhuman::context::tool_result_budget`]; this stage runs just +//! ahead of it in `agent_tool_exec::run_agent_tool_call`. +//! +//! Compressors: build/test logs, unified diffs, and JSON arrays (tabular; +//! large arrays additionally row-dropped with a reversible CCR offload — see +//! [`store`]). The system-prompt cache-aligner ([`cache_align`]) runs warn-only +//! from `ContextManager::build_system_prompt`. Every lossy path is recoverable +//! via the `retrieve_tool_output` tool, so it is safe under the always-on +//! default. +//! +//! **Search/grep output is intentionally not compacted** — see the router in +//! [`compact_tool_output`]. It's a completeness tool; structured match-dropping +//! does more harm than the tokens it saves. + +pub mod cache_align; +pub mod detect; +pub mod diff; +pub mod json_crusher; +pub mod logs; +pub mod signals; +pub mod store; + +#[cfg(test)] +mod demo; +#[cfg(test)] +mod measure; + +use detect::{hint_for_tool, resolve, ContentHint, ContentType}; +use std::fmt::Write as _; + +/// Outputs below this many bytes are never compressed — they're already cheap +/// and the structural compressors add overhead (markers) that can outweigh the +/// saving. Matches the spirit of the plan's `min_bytes_to_compress`. +pub const MIN_BYTES_TO_COMPRESS: usize = 2048; + +/// The CCR recovery tool's name (its `name()` in +/// `tools/impl/system/retrieve_tool_output.rs`). It has two cross-cutting +/// requirements, both keyed off this constant: +/// +/// 1. Its own output is **never compacted** ([`NEVER_COMPACT_TOOLS`]) — it +/// exists to return a previously-compacted original *in full*. +/// 2. It is **always advertised** to every agent regardless of `ToolScope`, +/// because compaction applies to every agent's tool output — so any agent +/// that sees a `retrieve_tool_output("…")` footer must actually be able to +/// call it (enforced in the tool-visibility filters). +pub const RECOVERY_TOOL_NAME: &str = "retrieve_tool_output"; + +/// Tools whose output must never be re-compacted. See [`RECOVERY_TOOL_NAME`]. +pub const NEVER_COMPACT_TOOLS: &[&str] = &[RECOVERY_TOOL_NAME]; + +/// Result of a single compressor: the compacted body plus whether any data was +/// actually dropped. Both kinds offload the original to CCR and carry a recovery +/// footer (see [`compact_tool_output`]); `lossy` only changes the wording — +/// "partial view" vs "faithful reformat" — so the model knows whether it's +/// missing data or just exact formatting. +pub struct Compacted { + pub text: String, + pub lossy: bool, +} + +impl Compacted { + /// All values preserved — only structure/formatting changed (e.g. the JSON + /// table). The exact original is still offered for recovery. + pub fn reformatted(text: String) -> Self { + Self { text, lossy: false } + } + /// Data was dropped — the original is offloaded so it stays recoverable. + pub fn lossy(text: String) -> Self { + Self { text, lossy: true } + } +} + +/// Compress a tool's output for the model context, routed by the tool name. +/// +/// Returns the (possibly) compacted string. Always falls back to the original +/// when: compaction is disabled, the output is small, the content type isn't +/// one we compress, or compression wouldn't shrink it. The result still flows +/// through the downstream byte budget, so this can only ever *help*. +/// +/// **Reversibility / honesty:** whenever a compressor drops data (`lossy`), the +/// full original is stashed in the [`store`] (CCR) and the returned text ends +/// with an explicit footer telling the model this is a partial view and how to +/// fetch the original via `retrieve_tool_output("")`. So the model is +/// never silently handed a truncated result, and nothing is unrecoverable. +pub fn compact_tool_output(content: String, tool_name: &str, enabled: bool) -> String { + if !enabled || content.len() < MIN_BYTES_TO_COMPRESS || NEVER_COMPACT_TOOLS.contains(&tool_name) + { + return content; + } + + let hint = hint_for_tool(tool_name); + // A Search hint is absolute: grep output is never compacted, even if its + // body happens to look diff-like — don't let `resolve` remap it to Diff. + let content_type = if matches!(hint, ContentHint::Search) { + ContentType::Search + } else { + resolve(hint, &content) + }; + + let compressed = match content_type { + // Search/grep output is deliberately NOT compacted. grep is a + // completeness tool — the agent runs it to find *every* call site — + // and dropping matches (even with a recovery footer) risks it acting + // on a partial set, with a relevance heuristic that doesn't apply to + // search results anyway. Large grep output is left to the downstream + // byte budget, which persists the full result to a `file_read`-able + // artifact rather than dropping it. See the design note in the PR. + ContentType::Search => None, + ContentType::Log => logs::compress(&content), + ContentType::Diff => diff::compress(&content), + ContentType::JsonArray => json_crusher::compress(&content), + // Plain text has no structural compressor (the ML text compressor is + // intentionally out of scope) — pass through to the byte budget. + ContentType::PlainText => None, + }; + + match compressed { + Some(c) if c.text.len() < content.len() => { + let mut out = c.text; + // Always offload the original and tell the model how to get it back. + // Lossy outputs are a partial view (data dropped); reformatted ones + // (the JSON table) keep every value but change layout — either way + // the exact original is one retrieve_tool_output call away. + let hash = store::offload(&content); + if c.lossy { + let _ = write!( + out, + "\n\n[compacted tool output — this is a PARTIAL view; the \ + full original ({} bytes) is available by calling \ + retrieve_tool_output(\"{hash}\")]", + content.len() + ); + } else { + let _ = write!( + out, + "\n\n[reformatted tool output — no data lost, but layout \ + changed; the exact original ({} bytes, e.g. raw JSON) is \ + available by calling retrieve_tool_output(\"{hash}\")]", + content.len() + ); + } + // The shrink check above ran on the compressed body; the recovery + // footer adds bytes, so re-check the final size and fall back to the + // original if the footer tipped it over (marginal inputs only). + if out.len() >= content.len() { + return content; + } + let ratio = 1.0 - (out.len() as f64 / content.len() as f64); + // `::log` is the logging crate (the sibling `logs` module shadows + // the bare `log` path inside this module). + ::log::debug!( + "[compaction] tool={tool_name} type={content_type:?} lossy={} in_bytes={} out_bytes={} ratio={ratio:.2}", + c.lossy, + content.len(), + out.len(), + ); + out + } + _ => content, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::fmt::Write as _; + + #[test] + fn disabled_is_passthrough() { + let big = "x".repeat(MIN_BYTES_TO_COMPRESS + 10); + assert_eq!(compact_tool_output(big.clone(), "grep", false), big); + } + + #[test] + fn small_output_passthrough() { + let small = "a.rs:1:hit\nb.rs:2:hit".to_string(); + assert_eq!(compact_tool_output(small.clone(), "grep", true), small); + } + + #[test] + fn search_output_is_not_compacted() { + // grep is a completeness tool — its output must pass through untouched + // even when large, so the agent never acts on a silently-dropped subset. + let mut s = String::from("80 match(es); scanned 2 file(s)\n"); + for i in 1..=40 { + let _ = writeln!( + s, + "src/a.rs:{i}:let value_{i} = compute_something_long_{i}();" + ); + } + for i in 1..=40 { + let _ = writeln!( + s, + "src/b.rs:{i}:fn helper_function_number_{i}() {{ /* body */ }}" + ); + } + assert!(s.len() >= MIN_BYTES_TO_COMPRESS); + let out = compact_tool_output(s.clone(), "grep", true); + assert_eq!(out, s, "search output must pass through unchanged"); + } + + #[test] + fn unknown_tool_plain_text_passthrough() { + let prose = "lorem ipsum ".repeat(400); // > MIN, but plain text + let out = compact_tool_output(prose.clone(), "some_tool", true); + assert_eq!(out, prose); + } + + /// Pull the CCR hash out of the retrieval footer the model is shown. + fn footer_hash(out: &str) -> Option<&str> { + out.split("retrieve_tool_output(\"") + .nth(1) + .and_then(|s| s.split('"').next()) + } + + #[test] + fn retrieval_returns_the_full_original_uncompacted() { + // End-to-end recovery: a large JSON result is compacted (lossy) and its + // original offloaded; the model reads the footer hash and calls + // retrieve_tool_output. That tool's output flows back through Stage 1a — + // and must NOT be re-compacted, or the agent could never see the full + // data it asked for. + let mut rows = Vec::new(); + for i in 0..120 { + rows.push(format!( + r#"{{"id":{i},"name":"account_{i}","email":"a{i}@ex.com","tier":"gold"}}"# + )); + } + let original = format!("[{}]", rows.join(",")); + + // 1. First pass compacts + offloads. + let compacted = compact_tool_output(original.clone(), "list_accounts", true); + assert!(compacted.contains("retrieve_tool_output(")); + let hash = footer_hash(&compacted).expect("footer hash"); + + // 2. The retrieve tool fetches the original from CCR. + let fetched = store::retrieve(hash).expect("CCR has it"); + assert_eq!(fetched, original); + + // 3. That fetched output passes through Stage 1a under the retrieve + // tool's name — and must come back byte-for-byte, NOT re-compacted. + let delivered = compact_tool_output(fetched, "retrieve_tool_output", true); + assert_eq!( + delivered, original, + "recovery must deliver the full original" + ); + assert!(!delivered.contains("partial view")); + } + + #[test] + fn every_lossy_output_tells_the_model_and_is_recoverable() { + // One representative input per lossy compressor. Each must (a) carry the + // explicit "partial view / retrieve_tool_output" footer, and (b) have + // its full original recoverable byte-for-byte from the CCR store — i.e. + // no information is actually lost, only deferred. (grep is excluded — + // search output is intentionally never compacted.) + let mut log = String::new(); + for i in 0..200 { + let _ = writeln!(log, " Compiling crate_{i} v0.{i}.0"); + } + let _ = writeln!(log, "error: aborting"); + let mut diff = String::from("diff --git a/x.rs b/x.rs\n@@ -1,60 +1,61 @@\n"); + for i in 0..50 { + let _ = writeln!( + diff, + " unchanged context line {i} carried along by the diff" + ); + } + let _ = writeln!(diff, "+changed"); + let mut jrows = Vec::new(); + for i in 0..120 { + jrows.push(format!( + r#"{{"id":{i},"name":"account_{i}","email":"a{i}@ex.com","tier":"gold"}}"# + )); + } + let json = format!("[{}]", jrows.join(",")); + + for (tool, input) in [ + ("run_tests", log), + ("read_diff", diff), + ("list_accounts", json), + ] { + let out = compact_tool_output(input.clone(), tool, true); + assert!(out.len() < input.len(), "{tool}: not compacted"); + // (a) the model is explicitly told this is a partial view. + assert!( + out.contains("PARTIAL view") && out.contains("retrieve_tool_output("), + "{tool}: missing retrieval footer:\n{out}" + ); + // (b) the full original is recoverable byte-for-byte. + let hash = footer_hash(&out).expect("footer has a hash"); + assert_eq!( + store::retrieve(hash).as_deref(), + Some(input.as_str()), + "{tool}: CCR did not round-trip" + ); + } + } + + #[test] + fn reformatted_table_preserves_values_and_offers_exact_recovery() { + // A JSON list under the row-drop threshold is reformatted, not dropped: + // every value is present (no "omitted"), it's framed as a faithful + // reformat (not a "partial view"), and the EXACT original JSON is + // recoverable via the retrieve footer — that's how the agent asks for + // exact bytes. + // 38 rows: above the 2048-byte floor, below the 40-row drop threshold. + let mut rows = Vec::new(); + for i in 0..38 { + rows.push(format!( + r#"{{"id":{i},"sku":"SKU-{i:05}","name":"widget number {i} in the catalog","warehouse":"east-region-1"}}"# + )); + } + let input = format!("[{}]", rows.join(",")); + assert!(input.len() >= MIN_BYTES_TO_COMPRESS); + let out = compact_tool_output(input.clone(), "list_inventory", true); + + assert!(out.len() < input.len(), "table should shrink"); + assert!(!out.contains("omitted"), "reformat ⇒ nothing dropped"); + // Framed as a faithful reformat, not a lossy partial view. + assert!(out.contains("no data lost"), "{out}"); + assert!(!out.contains("PARTIAL view"), "{out}"); + // Every row's identifying values survive the reformat. + for i in 0..38 { + assert!(out.contains(&format!("SKU-{i:05}")), "lost SKU {i}"); + assert!( + out.contains(&format!("widget number {i} in the catalog")), + "lost name {i}" + ); + } + // The agent can fetch the exact original JSON back, byte-for-byte. + let hash = footer_hash(&out).expect("reformat footer has a hash"); + assert_eq!(store::retrieve(hash).as_deref(), Some(input.as_str())); + } +} diff --git a/src/openhuman/agent/harness/compaction/signals.rs b/src/openhuman/agent/harness/compaction/signals.rs new file mode 100644 index 000000000..0ff777a6b --- /dev/null +++ b/src/openhuman/agent/harness/compaction/signals.rs @@ -0,0 +1,141 @@ +//! Shared importance signals for the compaction compressors. +//! +//! A small, deterministic keyword registry + per-line scorer used by +//! [`super::search`] and [`super::log`] to decide which lines to keep when a +//! tool output is over budget. No ML, no regex compilation cost on the hot +//! path beyond simple case-insensitive substring scans. +//! +//! Behavior is a clean-room port of headroom's `error_detection` priority +//! signals (Apache-2.0): error/fatal lines score highest, warnings next, +//! importance markers (security/TODO) a small bump, everything else baseline. + +/// Keywords that mark a hard failure. Matched case-insensitively as +/// substrings. Kept deliberately small and high-precision — a false positive +/// just means we keep a line we could have dropped, which is the safe +/// direction. +const ERROR_KEYWORDS: &[&str] = &[ + "error", + "fatal", + "panic", + "panicked", + "exception", + "traceback", + "failed", + "failure", + "segfault", + "assertion", + "abort", + "[error]", + "error:", +]; + +/// Keywords that mark a warning. Lower weight than errors. +const WARNING_KEYWORDS: &[&str] = &["warning", "warn:", "[warn]", "deprecated"]; + +/// Keywords that bump importance regardless of severity — things an agent +/// almost always wants to see even in a truncated view. +const IMPORTANCE_KEYWORDS: &[&str] = &[ + "security", + "vulnerability", + "critical", + "todo", + "fixme", + "denied", + "unauthorized", + "forbidden", +]; + +/// Score weights. Higher = more likely to survive truncation. +pub const SCORE_ERROR: f32 = 1.0; +pub const SCORE_WARNING: f32 = 0.6; +pub const SCORE_IMPORTANCE: f32 = 0.4; +pub const SCORE_BASELINE: f32 = 0.1; + +/// True if any error keyword appears in `text` (case-insensitive). Cheap +/// pre-check used to decide whether a blob is worth the log compressor. +pub fn has_error_indicators(text: &str) -> bool { + let lower = text.to_ascii_lowercase(); + ERROR_KEYWORDS.iter().any(|kw| lower.contains(kw)) +} + +/// Importance score for a single line in `[0.0, 1.0]`. Errors dominate, +/// then warnings, then importance markers; a plain line gets the baseline so +/// ordering is stable and "keep highest N" never discards everything. +pub fn line_score(line: &str) -> f32 { + let lower = line.to_ascii_lowercase(); + let mut score = SCORE_BASELINE; + if ERROR_KEYWORDS.iter().any(|kw| lower.contains(kw)) { + score = score.max(SCORE_ERROR); + } + if WARNING_KEYWORDS.iter().any(|kw| lower.contains(kw)) { + score = score.max(SCORE_WARNING); + } + if IMPORTANCE_KEYWORDS.iter().any(|kw| lower.contains(kw)) { + score = score.max(SCORE_IMPORTANCE); + } + score +} + +/// Classify a line's severity for the log compressor's bucketing. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Severity { + Error, + Warning, + Other, +} + +/// Bucket a line into [`Severity`]. Used by [`super::log`] to keep error and +/// warning lines under separate caps. +pub fn severity(line: &str) -> Severity { + let lower = line.to_ascii_lowercase(); + if ERROR_KEYWORDS.iter().any(|kw| lower.contains(kw)) { + Severity::Error + } else if WARNING_KEYWORDS.iter().any(|kw| lower.contains(kw)) { + Severity::Warning + } else { + Severity::Other + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn errors_score_highest() { + assert_eq!(line_score("FATAL: connection refused"), SCORE_ERROR); + assert_eq!(line_score("thread panicked at 'boom'"), SCORE_ERROR); + assert!(line_score("error: mismatched types") >= SCORE_ERROR); + } + + #[test] + fn warnings_below_errors_above_baseline() { + let w = line_score("warning: unused variable"); + assert!(w < SCORE_ERROR); + assert!(w > SCORE_BASELINE); + } + + #[test] + fn plain_line_is_baseline() { + assert_eq!(line_score(" Compiling foo v0.1.0"), SCORE_BASELINE); + } + + #[test] + fn importance_markers_bump() { + assert!(line_score("TODO: handle retry") > SCORE_BASELINE); + assert!(line_score("potential security issue here") > SCORE_BASELINE); + } + + #[test] + fn severity_buckets() { + assert_eq!(severity("error[E0382]: borrow"), Severity::Error); + assert_eq!(severity("warning: deprecated"), Severity::Warning); + assert_eq!(severity("running 12 tests"), Severity::Other); + } + + #[test] + fn has_error_indicators_detects() { + assert!(has_error_indicators("test result: FAILED")); + assert!(!has_error_indicators("all good, 12 passed")); + } +} diff --git a/src/openhuman/agent/harness/compaction/store.rs b/src/openhuman/agent/harness/compaction/store.rs new file mode 100644 index 000000000..4a3c07c4c --- /dev/null +++ b/src/openhuman/agent/harness/compaction/store.rs @@ -0,0 +1,135 @@ +//! CCR — Compress-Cache-Retrieve store. +//! +//! When a compressor drops data (lossy paths), it stows the original here +//! keyed by a short content hash and emits a `retrieve_tool_output("")` +//! sentinel in the compacted text. The agent can call the +//! `retrieve_tool_output` tool to get the original back on demand — so even +//! aggressive compaction stays reversible and is safe under the always-on +//! default. +//! +//! Process-global and bounded: a fixed-capacity FIFO so a long session can't +//! grow it without bound. Keyed by content hash, so re-offloading identical +//! content is idempotent (the model sees a stable hash). Originals are not +//! persisted to disk — retrieval is best-effort within the session; an evicted +//! entry simply reports "no longer available", which is strictly better than +//! the pre-CCR behaviour (the data was gone the moment it was truncated). + +use sha2::{Digest, Sha256}; +use std::collections::{HashMap, VecDeque}; +use std::sync::{Mutex, OnceLock}; + +/// Max originals retained. ~256 large tool outputs is plenty for a session's +/// recent history while bounding worst-case memory. +const MAX_ENTRIES: usize = 256; +/// Bytes of the SHA-256 digest used for the key (→ 32 hex chars). Wide enough +/// that (a) collisions are infeasible and (b) the hash doubles as an +/// unguessable capability token — a session can only retrieve content whose +/// hash it was shown, so the process-global store can't be brute-force probed +/// across sessions. (Full per-session key namespacing is a tracked follow-up.) +const HASH_BYTES: usize = 16; + +#[derive(Default)] +struct Inner { + map: HashMap, + order: VecDeque, +} + +impl Inner { + /// Insert `content` under `hash` (idempotent) and FIFO-evict down to + /// [`MAX_ENTRIES`]. Pulled out of [`offload`] so the eviction policy can be + /// unit-tested on a local instance without touching the process-global + /// store (which would otherwise race other tests sharing it). + fn insert(&mut self, hash: String, content: String) { + if self.map.contains_key(&hash) { + return; + } + self.map.insert(hash.clone(), content); + self.order.push_back(hash); + while self.order.len() > MAX_ENTRIES { + if let Some(evicted) = self.order.pop_front() { + self.map.remove(&evicted); + } + } + } +} + +fn global() -> &'static Mutex { + static STORE: OnceLock> = OnceLock::new(); + STORE.get_or_init(|| Mutex::new(Inner::default())) +} + +/// Stash `content` and return its short hash. Idempotent for identical content. +pub fn offload(content: &str) -> String { + let hash = short_hash(content); + global() + .lock() + .unwrap_or_else(|p| p.into_inner()) + .insert(hash.clone(), content.to_string()); + hash +} + +/// Retrieve a previously-offloaded original by hash, if still cached. +pub fn retrieve(hash: &str) -> Option { + global() + .lock() + .unwrap_or_else(|p| p.into_inner()) + .map + .get(hash) + .cloned() +} + +/// Short hex content hash used as the CCR key. +pub fn short_hash(content: &str) -> String { + let mut hasher = Sha256::new(); + hasher.update(content.as_bytes()); + let digest = hasher.finalize(); + hex::encode(&digest[..HASH_BYTES]) +} + +#[cfg(test)] +mod tests { + use super::*; + + // Round-trip tests use globally-unique content and collectively stay well + // under MAX_ENTRIES, so they never evict each other even under parallel + // execution. The eviction policy is exercised on a *local* `Inner` below so + // it can't clobber the shared store other tests depend on. + + #[test] + fn round_trips() { + let original = "ccr round-trip unique payload ".repeat(50); + let hash = offload(&original); + assert_eq!(hash.len(), HASH_BYTES * 2); + assert_eq!(retrieve(&hash).as_deref(), Some(original.as_str())); + } + + #[test] + fn idempotent_hash() { + let a = offload("ccr idempotent unique payload content here"); + let b = offload("ccr idempotent unique payload content here"); + assert_eq!(a, b); + } + + #[test] + fn missing_hash_is_none() { + // A 32-hex hash that no test content maps to. + assert!(retrieve("ffffffffffffffffffffffffffffffff").is_none()); + } + + #[test] + fn eviction_bounds_size() { + // Exercise the FIFO eviction on a local instance — no shared state. + let mut inner = Inner::default(); + for i in 0..(MAX_ENTRIES + 50) { + inner.insert(format!("hash-{i}"), format!("content-{i}")); + } + assert!(inner.map.len() <= MAX_ENTRIES, "size bounded"); + assert!(!inner.map.contains_key("hash-0"), "oldest entry evicted"); + assert!( + inner + .map + .contains_key(&format!("hash-{}", MAX_ENTRIES + 49)), + "newest entry retained" + ); + } +} diff --git a/src/openhuman/agent/harness/mod.rs b/src/openhuman/agent/harness/mod.rs index 57c910b82..87ed9ade3 100644 --- a/src/openhuman/agent/harness/mod.rs +++ b/src/openhuman/agent/harness/mod.rs @@ -21,6 +21,7 @@ pub mod archivist; pub(crate) mod builtin_definitions; +pub(crate) mod compaction; mod credentials; pub mod definition; pub(crate) mod definition_loader; diff --git a/src/openhuman/agent/harness/session/agent_tool_exec.rs b/src/openhuman/agent/harness/session/agent_tool_exec.rs index d71634ce3..18480a8b9 100644 --- a/src/openhuman/agent/harness/session/agent_tool_exec.rs +++ b/src/openhuman/agent/harness/session/agent_tool_exec.rs @@ -38,6 +38,9 @@ pub(super) struct AgentToolExecCtx<'a> { pub agent_definition_id: &'a str, pub prefer_markdown: bool, pub budget_bytes: usize, + /// Whether Stage 1a (native content-aware compaction) runs before the + /// byte budget. Sourced from `ContextManager::compaction_enabled`. + pub compaction_enabled: bool, pub artifact_store: Option<&'a ToolResultArtifactStore>, } @@ -226,6 +229,16 @@ pub(super) async fn run_agent_tool_call( (format!("Unknown tool: {}", call.name), false) }; + // Stage 1a — content-aware compaction. Runs before the byte budget on the + // fresh tool output (never sent to the backend yet, so it's cache-safe like + // the budget below). Routes by tool name; only ever shrinks, otherwise + // passes the original through. See `agent::harness::compaction`. + let raw_result = crate::openhuman::agent::harness::compaction::compact_tool_output( + raw_result, + &call.name, + ctx.compaction_enabled, + ); + // Per-result byte budget — the only cache-safe reduction stage (the full // body has never been sent to the backend). Oversized outputs are persisted // into the action workspace when possible, with truncation as fallback. diff --git a/src/openhuman/agent/harness/session/builder/builder_tests.rs b/src/openhuman/agent/harness/session/builder/builder_tests.rs index 768c11d79..5e4bdd967 100644 --- a/src/openhuman/agent/harness/session/builder/builder_tests.rs +++ b/src/openhuman/agent/harness/session/builder/builder_tests.rs @@ -1,6 +1,8 @@ //! Tests for the builder module — dedup_visible_tool_specs and related logic. -use super::{dedup_visible_tool_specs, should_synthesize_delegation_tools}; +use super::{ + dedup_visible_tool_specs, ensure_recovery_tool_visible, should_synthesize_delegation_tools, +}; use crate::openhuman::tools::ToolSpec; use serde_json::json; @@ -12,6 +14,36 @@ fn spec(name: &str) -> ToolSpec { } } +#[test] +fn recovery_tool_joins_a_named_allowlist() { + use crate::openhuman::agent::harness::compaction::RECOVERY_TOOL_NAME; + use std::collections::HashSet; + + // A curated Named-scope allowlist gains retrieve_tool_output as a *real* + // member, so the policy session, advertised specs, and the run-time + // visible-name gate (all driven by this set) make a compaction footer + // actionable. + let mut visible: HashSet = ["file_read".to_string(), "grep".to_string()] + .into_iter() + .collect(); + ensure_recovery_tool_visible(&mut visible); + assert!( + visible.contains(RECOVERY_TOOL_NAME), + "recovery tool must join: {visible:?}" + ); + assert!(visible.contains("file_read")); +} + +#[test] +fn empty_allowlist_stays_empty() { + use std::collections::HashSet; + // Empty == "no filter" (all tools visible) AND the deliberately tool-less + // Named([]) case — both must stay empty so the invariant holds. + let mut visible: HashSet = HashSet::new(); + ensure_recovery_tool_visible(&mut visible); + assert!(visible.is_empty(), "empty allowlist must not gain a tool"); +} + #[test] fn drops_duplicates_first_wins() { // Real-world collision: researcher's `delegate_name = "research"` diff --git a/src/openhuman/agent/harness/session/builder/factory.rs b/src/openhuman/agent/harness/session/builder/factory.rs index 585ab1f59..f3d0d2680 100644 --- a/src/openhuman/agent/harness/session/builder/factory.rs +++ b/src/openhuman/agent/harness/session/builder/factory.rs @@ -872,6 +872,16 @@ impl Agent { .map(|t| t.name().to_string()) .collect(), }; + // Compaction applies to every agent's tool output, so the CCR recovery + // tool must be a *real* member of any non-empty allowlist — this is the + // single source of truth that the policy session, advertised specs, and + // the run-time visible-name gate all consume, so adding it here makes a + // `retrieve_tool_output("…")` footer actionable for Named-scope agents + // (e.g. the orchestrator's curated list). An empty set already means + // "no filter", so it needs nothing. Added BEFORE the disallow filter + // below so an agent that explicitly disallows it still has it removed. + super::ensure_recovery_tool_visible(&mut visible); + if let Some(def) = target_def { if !def.disallowed_tools.is_empty() { match &def.tools { diff --git a/src/openhuman/agent/harness/session/builder/mod.rs b/src/openhuman/agent/harness/session/builder/mod.rs index 88e164f68..0a3175e6c 100644 --- a/src/openhuman/agent/harness/session/builder/mod.rs +++ b/src/openhuman/agent/harness/session/builder/mod.rs @@ -66,6 +66,19 @@ pub(super) fn visible_tool_specs_for_policy( .collect() } +/// Ensure the CCR recovery tool (`retrieve_tool_output`) is a member of a +/// non-empty visibility allowlist. Compaction runs on every agent's tool +/// output, so any agent with a curated `ToolScope::Named` list must still be +/// able to act on a `retrieve_tool_output("…")` footer. An empty set already +/// means "no filter" (all tools visible), so it is left untouched — including +/// the deliberately tool-less `Named([])` case, which must stay tool-less. +pub(super) fn ensure_recovery_tool_visible(visible: &mut std::collections::HashSet) { + if !visible.is_empty() { + visible + .insert(crate::openhuman::agent::harness::compaction::RECOVERY_TOOL_NAME.to_string()); + } +} + pub(super) fn should_synthesize_delegation_tools(def: &AgentDefinition) -> bool { match &def.tools { ToolScope::Wildcard => !def.subagents.is_empty(), diff --git a/src/openhuman/agent/harness/session/turn/core.rs b/src/openhuman/agent/harness/session/turn/core.rs index 8decfa0b7..ddbeb21ea 100644 --- a/src/openhuman/agent/harness/session/turn/core.rs +++ b/src/openhuman/agent/harness/session/turn/core.rs @@ -483,6 +483,7 @@ impl Agent { agent_definition_id: self.agent_definition_id.clone(), prefer_markdown: self.context.prefer_markdown_tool_output(), budget_bytes: self.context.tool_result_budget_bytes(), + compaction_enabled: self.context.compaction_enabled(), artifact_store: artifact_store.clone(), should_send_specs: self.tool_dispatcher.should_send_tool_specs(), advertised_specs: self.visible_tool_specs.as_ref().clone(), diff --git a/src/openhuman/agent/harness/session/turn/tools.rs b/src/openhuman/agent/harness/session/turn/tools.rs index 735f0d9e0..e1dc4fd2d 100644 --- a/src/openhuman/agent/harness/session/turn/tools.rs +++ b/src/openhuman/agent/harness/session/turn/tools.rs @@ -57,6 +57,7 @@ impl Agent { agent_definition_id: &self.agent_definition_id, prefer_markdown: self.context.prefer_markdown_tool_output(), budget_bytes: self.context.tool_result_budget_bytes(), + compaction_enabled: self.context.compaction_enabled(), artifact_store: Some(&artifact_store), }; agent_tool_exec::run_agent_tool_call(&ctx, &progress, call, iteration).await diff --git a/src/openhuman/agent/harness/session/turn_engine_adapter.rs b/src/openhuman/agent/harness/session/turn_engine_adapter.rs index 518b8d729..4db311371 100644 --- a/src/openhuman/agent/harness/session/turn_engine_adapter.rs +++ b/src/openhuman/agent/harness/session/turn_engine_adapter.rs @@ -99,6 +99,9 @@ pub(super) struct AgentToolSource { pub agent_definition_id: String, pub prefer_markdown: bool, pub budget_bytes: usize, + /// Stage 1a kill-switch. Constant for the session, so (unlike the tool + /// surface) it is set once at construction and never re-synced. + pub compaction_enabled: bool, pub artifact_store: Option, pub should_send_specs: bool, pub advertised_specs: Vec, @@ -141,6 +144,7 @@ impl ToolSource for AgentToolSource { agent_definition_id: &self.agent_definition_id, prefer_markdown: self.prefer_markdown, budget_bytes: self.budget_bytes, + compaction_enabled: self.compaction_enabled, artifact_store: self.artifact_store.as_ref(), }; let (exec_result, record) = diff --git a/src/openhuman/agent/harness/subagent_runner/tool_prep.rs b/src/openhuman/agent/harness/subagent_runner/tool_prep.rs index cb975bc21..83d0da55a 100644 --- a/src/openhuman/agent/harness/subagent_runner/tool_prep.rs +++ b/src/openhuman/agent/harness/subagent_runner/tool_prep.rs @@ -143,6 +143,15 @@ pub(crate) fn filter_tool_indices( if disallowed_tool_matches(disallowed, name) { return false; } + // The CCR recovery tool is advertised to any agent that has a tool + // surface — compaction applies to its tool output, so the retrieve + // footer must be actionable regardless of scope/skill filters (an + // explicit `disallow` above still wins). A deliberately tool-less + // agent (`Named([])`, e.g. the payload summarizer) runs no tools, + // produces no compacted output, and so stays tool-less. + if name == crate::openhuman::agent::harness::compaction::RECOVERY_TOOL_NAME { + return !matches!(scope, ToolScope::Named(allowed) if allowed.is_empty()); + } if let Some(prefix) = skill_prefix.as_deref() { if !name.starts_with(prefix) { return false; @@ -180,6 +189,72 @@ mod tests { } } +#[cfg(test)] +mod recovery_visibility_tests { + use super::*; + use crate::openhuman::agent::harness::compaction::RECOVERY_TOOL_NAME; + use crate::openhuman::tools::{CurrentTimeTool, RetrieveToolOutputTool}; + + fn tools() -> Vec> { + vec![ + Box::new(CurrentTimeTool::new()), + Box::new(RetrieveToolOutputTool::new()), + ] + } + + fn names(idx: &[usize], tools: &[Box]) -> Vec { + idx.iter().map(|&i| tools[i].name().to_string()).collect() + } + + #[test] + fn named_scope_still_includes_recovery_tool() { + let t = tools(); + // Named scope allow-lists only current_time — recovery tool not listed. + let idx = filter_tool_indices( + &t, + &ToolScope::Named(vec!["current_time".into()]), + &[], + None, + ); + let got = names(&idx, &t); + assert!(got.contains(&"current_time".to_string())); + assert!( + got.contains(&RECOVERY_TOOL_NAME.to_string()), + "recovery tool must survive Named scope: {got:?}" + ); + } + + #[test] + fn tool_less_agent_stays_tool_less() { + // A deliberately tool-less agent (e.g. the payload summarizer, + // ToolScope::Named([])) runs no tools and produces no compacted output, + // so it must NOT be handed the recovery tool — it stays empty. + let t = tools(); + let idx = filter_tool_indices(&t, &ToolScope::Named(vec![]), &[], None); + assert!(idx.is_empty(), "empty scope must yield zero tools: {idx:?}"); + } + + #[test] + fn skill_filter_still_includes_recovery_tool() { + let t = tools(); + // A skill-restricted subagent (only `foo__*` tools) must still get it. + let idx = filter_tool_indices(&t, &ToolScope::Wildcard, &[], Some("foo")); + assert!(names(&idx, &t).contains(&RECOVERY_TOOL_NAME.to_string())); + } + + #[test] + fn explicit_disallow_still_wins() { + let t = tools(); + let idx = filter_tool_indices( + &t, + &ToolScope::Wildcard, + &[RECOVERY_TOOL_NAME.to_string()], + None, + ); + assert!(!names(&idx, &t).contains(&RECOVERY_TOOL_NAME.to_string())); + } +} + // ── Prompt loading ────────────────────────────────────────────────────── /// Resolve a [`PromptSource`] to its raw markdown body. Inline sources diff --git a/src/openhuman/config/schema/context.rs b/src/openhuman/config/schema/context.rs index 66291991e..79ef0b482 100644 --- a/src/openhuman/config/schema/context.rs +++ b/src/openhuman/config/schema/context.rs @@ -103,6 +103,21 @@ pub struct ContextConfig { /// downstream consumer expects strict JSON tool output. #[serde(default = "default_true")] pub prefer_markdown_tool_output: bool, + + /// Master switch for native tool-output compaction (Stage 1a). When + /// `true` (the default), large structured tool outputs (build/test logs, + /// diffs, JSON arrays) are content-aware compressed in + /// `Agent::execute_tool_call` *before* the [`Self::tool_result_budget_bytes`] + /// byte cap and before they enter history. The compression never drops the + /// first/last/high-signal lines and only ever shrinks output, so it is on + /// by default. + /// + /// This is invisible infrastructure (like microcompact/autocompact): no + /// user-facing UI. The only reason to flip it off is a support / debugging + /// / A/B bisect, via config or the `OPENHUMAN_COMPACTION=0` env override. + /// See `compaction-plan.md`. + #[serde(default = "default_true")] + pub compaction_enabled: bool, } fn default_enabled() -> bool { @@ -146,6 +161,7 @@ impl Default for ContextConfig { session_memory: SessionMemoryConfig::default(), summarizer_model: None, prefer_markdown_tool_output: default_true(), + compaction_enabled: default_true(), } } } diff --git a/src/openhuman/config/schema/load/env_overlay.rs b/src/openhuman/config/schema/load/env_overlay.rs index cc2a6ee4d..779f7ae27 100644 --- a/src/openhuman/config/schema/load/env_overlay.rs +++ b/src/openhuman/config/schema/load/env_overlay.rs @@ -787,6 +787,19 @@ impl Config { self.context.tool_result_budget_bytes = n; } } + // Kill-switch for native tool-output compaction (Stage 1a). On by + // default; `OPENHUMAN_COMPACTION=0` disables it for a support/A-B + // bisect. Accepts the canonical short name and the namespaced form. + if let Some(flag) = env + .get("OPENHUMAN_COMPACTION") + .or_else(|| env.get("OPENHUMAN_CONTEXT_COMPACTION_ENABLED")) + { + match flag.trim().to_ascii_lowercase().as_str() { + "1" | "true" | "yes" | "on" => self.context.compaction_enabled = true, + "0" | "false" | "no" | "off" => self.context.compaction_enabled = false, + _ => {} + } + } if let Some(model) = env.get("OPENHUMAN_CONTEXT_SUMMARIZER_MODEL") { let model = model.trim(); if !model.is_empty() { diff --git a/src/openhuman/config/schema/load_tests.rs b/src/openhuman/config/schema/load_tests.rs index 83461b049..03ab70dfe 100644 --- a/src/openhuman/config/schema/load_tests.rs +++ b/src/openhuman/config/schema/load_tests.rs @@ -984,6 +984,30 @@ fn env_overlay_context_tool_result_budget_env_suppresses_legacy_migration() { ); } +#[test] +fn env_overlay_compaction_default_on_and_kill_switch() { + // Default is on. + assert!(Config::default().context.compaction_enabled); + + // `OPENHUMAN_COMPACTION=0` disables it. + let mut cfg = Config::default(); + cfg.apply_env_overlay_with(&HashMapEnv::new().with("OPENHUMAN_COMPACTION", "0")); + assert!(!cfg.context.compaction_enabled); + + // Truthy re-enables; the namespaced alias works too. + let mut cfg = Config::default(); + cfg.context.compaction_enabled = false; + cfg.apply_env_overlay_with( + &HashMapEnv::new().with("OPENHUMAN_CONTEXT_COMPACTION_ENABLED", "on"), + ); + assert!(cfg.context.compaction_enabled); + + // Garbage is ignored (leaves the prior value untouched). + let mut cfg = Config::default(); + cfg.apply_env_overlay_with(&HashMapEnv::new().with("OPENHUMAN_COMPACTION", "maybe")); + assert!(cfg.context.compaction_enabled); +} + #[test] fn env_overlay_context_tool_result_budget_legacy_migration_when_env_absent() { // Env absent, context at default, agent customised → agent value copies forward. diff --git a/src/openhuman/context/manager.rs b/src/openhuman/context/manager.rs index 6cc1fbfce..8d131531b 100644 --- a/src/openhuman/context/manager.rs +++ b/src/openhuman/context/manager.rs @@ -120,6 +120,11 @@ pub struct ContextManager { /// markdown instead of JSON — significantly cheaper in the model /// context window. See [`ContextConfig::prefer_markdown_tool_output`]. prefer_markdown_tool_output: bool, + /// When `true`, native tool-output compaction (Stage 1a) runs in + /// `Agent::execute_tool_call` before the byte cap. On by default; the + /// kill-switch lives here so every caller reads one source of truth. + /// See [`ContextConfig::compaction_enabled`]. + compaction_enabled: bool, } impl ContextManager { @@ -161,6 +166,7 @@ impl ContextManager { enabled: config.enabled, tool_result_budget_bytes: config.tool_result_budget_bytes, prefer_markdown_tool_output: config.prefer_markdown_tool_output, + compaction_enabled: config.compaction_enabled, } } @@ -178,6 +184,13 @@ impl ContextManager { self.tool_result_budget_bytes } + /// Whether native tool-output compaction (Stage 1a) is enabled. Agents + /// read this when a tool returns to decide whether to content-aware + /// compress the result before the byte cap and before it enters history. + pub fn compaction_enabled(&self) -> bool { + self.compaction_enabled + } + // ─── Budget tracking ────────────────────────────────────────── /// Feed the latest provider [`UsageInfo`] into the guard + the @@ -248,7 +261,9 @@ impl ContextManager { /// the inference backend's prefix cache picks up the stable prefix /// automatically, so no boundary marker is emitted. pub fn build_system_prompt(&self, ctx: &PromptContext<'_>) -> Result { - self.default_prompt_builder.build(ctx) + let prompt = self.default_prompt_builder.build(ctx)?; + self.warn_if_cache_unstable(&prompt); + Ok(prompt) } /// Assemble the system prompt via a caller-supplied builder. @@ -263,7 +278,19 @@ impl ContextManager { builder: &SystemPromptBuilder, ctx: &PromptContext<'_>, ) -> Result { - builder.build(ctx) + let prompt = builder.build(ctx)?; + self.warn_if_cache_unstable(&prompt); + Ok(prompt) + } + + /// Cache-aligner (Stage 1a sibling, warn-only): flag volatile tokens in + /// the cache-hot system prompt that would silently break the provider + /// KV-cache prefix. Never mutates the prompt. Gated on the compaction + /// kill-switch so disabling compaction also silences this diagnostic. + fn warn_if_cache_unstable(&self, prompt: &str) { + if self.compaction_enabled { + crate::openhuman::agent::harness::compaction::cache_align::warn_if_volatile(prompt); + } } // ─── Reduction ───────────────────────────────────────────────── diff --git a/src/openhuman/tools/impl/system/mod.rs b/src/openhuman/tools/impl/system/mod.rs index 75b8bc93b..cb22d0c72 100644 --- a/src/openhuman/tools/impl/system/mod.rs +++ b/src/openhuman/tools/impl/system/mod.rs @@ -9,6 +9,7 @@ mod npm_exec; mod proxy_config; mod pushover; mod resolve_time; +mod retrieve_tool_output; mod schedule; mod shell; mod tool_stats; @@ -29,6 +30,7 @@ pub use npm_exec::NpmExecTool; pub use proxy_config::ProxyConfigTool; pub use pushover::PushoverTool; pub use resolve_time::ResolveTimeTool; +pub use retrieve_tool_output::RetrieveToolOutputTool; pub use schedule::ScheduleTool; pub use shell::ShellTool; pub use tool_stats::ToolStatsTool; diff --git a/src/openhuman/tools/impl/system/retrieve_tool_output.rs b/src/openhuman/tools/impl/system/retrieve_tool_output.rs new file mode 100644 index 000000000..decb6589d --- /dev/null +++ b/src/openhuman/tools/impl/system/retrieve_tool_output.rs @@ -0,0 +1,115 @@ +//! Tool: retrieve_tool_output — fetch the original of a compacted tool result. +//! +//! Native tool-output compaction (Stage 1a) may replace a large tool result +//! with a compacted view and a `retrieve_tool_output("")` sentinel, +//! stashing the original in the CCR store +//! ([`crate::openhuman::agent::harness::compaction::store`]). This tool hands +//! the original back on demand, so even lossy compaction stays reversible. +//! +//! Read-only, no side effects, no path/network access. + +use crate::openhuman::tools::traits::{PermissionLevel, Tool, ToolResult}; +use async_trait::async_trait; +use serde_json::json; + +pub struct RetrieveToolOutputTool; + +impl RetrieveToolOutputTool { + pub fn new() -> Self { + Self + } +} + +impl Default for RetrieveToolOutputTool { + fn default() -> Self { + Self::new() + } +} + +#[async_trait] +impl Tool for RetrieveToolOutputTool { + fn name(&self) -> &str { + "retrieve_tool_output" + } + + fn description(&self) -> &str { + "Retrieve the full, original text of a tool result that was compacted to \ + save context. When a tool output shows a marker like \ + `retrieve_tool_output(\"a1b2c3d4e5f6\")`, call this with that hash to get \ + the complete original back. Use it only when you actually need the dropped \ + detail — the compacted view is usually enough." + } + + fn parameters_schema(&self) -> serde_json::Value { + json!({ + "type": "object", + "properties": { + "hash": { + "type": "string", + "description": "The hash from a retrieve_tool_output(\"…\") marker." + } + }, + "required": ["hash"] + }) + } + + fn permission_level(&self) -> PermissionLevel { + PermissionLevel::ReadOnly + } + + async fn execute(&self, args: serde_json::Value) -> anyhow::Result { + let hash = args + .get("hash") + .and_then(|v| v.as_str()) + .map(str::trim) + .filter(|s| !s.is_empty()); + let Some(hash) = hash else { + return Ok(ToolResult::error( + "retrieve_tool_output: missing required 'hash' argument".to_string(), + )); + }; + + match crate::openhuman::agent::harness::compaction::store::retrieve(hash) { + Some(original) => { + log::debug!( + "[compaction][ccr] retrieved hash={} bytes={}", + hash, + original.len() + ); + Ok(ToolResult::success(original)) + } + None => Ok(ToolResult::error(format!( + "retrieve_tool_output: no cached original for hash '{hash}' \ + (it may have been evicted; re-run the tool to regenerate it)" + ))), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::agent::harness::compaction::store; + + #[tokio::test] + async fn retrieves_offloaded_original() { + let original = "ORIGINAL PAYLOAD ".repeat(20); + let hash = store::offload(&original); + let tool = RetrieveToolOutputTool::new(); + let res = tool.execute(json!({ "hash": hash })).await.unwrap(); + assert!(!res.is_error); + assert_eq!(res.output(), original); + } + + #[tokio::test] + async fn missing_hash_is_error() { + let tool = RetrieveToolOutputTool::new(); + let res = tool + .execute(json!({ "hash": "deadbeefcafe" })) + .await + .unwrap(); + assert!(res.is_error); + let res2 = tool.execute(json!({})).await.unwrap(); + assert!(res2.is_error); + } +} diff --git a/src/openhuman/tools/ops.rs b/src/openhuman/tools/ops.rs index ac3b7b28b..ea18f1fd0 100644 --- a/src/openhuman/tools/ops.rs +++ b/src/openhuman/tools/ops.rs @@ -198,6 +198,10 @@ pub fn all_tools_with_runtime( Box::new(RunWorkflowTool::new().with_skill_allowlist(skill_allowlist.cloned())), Box::new(AwaitWorkflowTool::new()), Box::new(CurrentTimeTool::new()), + // Reversibility for native tool-output compaction (Stage 1a): when a + // large result is compacted with a `retrieve_tool_output("")` + // marker, this hands the original back from the CCR store on demand. + Box::new(RetrieveToolOutputTool::new()), // Deterministic time-expression → timestamp resolver. `current_time` // only returns *now*, leaving the model to do epoch arithmetic by hand // (a real incident had an agent compute "24h ago" ~10 months off, then