mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
* feat(agent): summarizer sub-agent compresses oversized tool results Issue #574. Before this change, tool results larger than a few KB landed verbatim in orchestrator history, burning context budget on raw JSON/HTML/file dumps. The only guardrail was tool_result_budget_bytes which hard-truncated mid-payload, dropping everything past the cut. This PR introduces a dedicated `summarizer` sub-agent (model.hint = "summarization") that the runtime automatically dispatches whenever a tool result exceeds summarizer_payload_threshold_bytes (default 100 KB). The summarizer compresses the payload per an extraction contract that preserves identifiers and key facts, and the compressed summary replaces the raw payload before it enters agent history. Payloads above summarizer_max_payload_bytes (default 5 MB) skip summarization and fall through to the existing truncation path — paying for an LLM call on a 5 MB blob is counterproductive. Scoped to the orchestrator session only. Welcome, skills_agent, researcher, planner, and every other typed sub-agent get None and their tool results are untouched. Gated in build_session_agent_inner by checking agent_id == "orchestrator". Instrumented at both tool-loop paths: - run_tool_call_loop in tool_loop.rs (event-bus/channels path) - Agent::execute_tool_call in session/turn.rs (web channel path via run_single) Both paths call into the same `PayloadSummarizer::maybe_summarize` trait so the threshold check, circuit breaker (3 consecutive failures disables for the session), and sub-agent dispatch policy live in exactly one place (agent/harness/payload_summarizer.rs). The summarizer sub-agent is runtime-dispatched only — it is NOT exposed as a delegation tool to the orchestrator's LLM. It is listed in the orchestrator's `subagents = [...]` for explicit registration, but `collect_orchestrator_tools` filters out the `summarizer` id and never synthesises a `delegate_summarizer` tool. Files: * NEW: src/openhuman/agent/agents/summarizer/{agent.toml,prompt.md} * NEW: src/openhuman/agent/harness/payload_summarizer.rs (trait + SubagentPayloadSummarizer impl + circuit breaker + tests) * src/openhuman/agent/agents/mod.rs — register summarizer in BUILTINS * src/openhuman/agent/agents/orchestrator/agent.toml — list summarizer in subagents (runtime-only, no LLM delegation tool) * src/openhuman/agent/harness/mod.rs — declare module * src/openhuman/agent/harness/builtin_definitions.rs — expect summarizer in `expected_builtin_ids_are_present` * src/openhuman/agent/harness/tool_loop.rs — thread Option<&dyn PayloadSummarizer> into run_tool_call_loop + agent_turn, intercept at the tool-execution success site, plus a new integration test using a MockSummarizer * src/openhuman/agent/harness/tests.rs — thread None through the existing run_tool_call_loop callsites * src/openhuman/agent/harness/session/turn.rs — same interception in Agent::execute_tool_call * src/openhuman/agent/harness/session/types.rs — payload_summarizer field on Agent and AgentBuilder * src/openhuman/agent/harness/session/builder.rs — .payload_summarizer() setter + orchestrator-only construction in build_session_agent_inner * src/openhuman/agent/bus.rs — pass None for the bus path * src/openhuman/config/schema/context.rs — summarizer_payload_threshold_bytes + summarizer_max_payload_bytes on ContextConfig * src/openhuman/tools/orchestrator_tools.rs — filter out the summarizer id from delegation-tool synthesis Tests: unit tests in payload_summarizer pin the pass-through rules (below threshold, above max cap, breaker tripped) and the prompt construction. Integration test in tool_loop::tests uses a MockSummarizer to verify the interception wires through end-to-end. Closes tinyhumansai/openhuman#574. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * feat(tools): csv_export tool + skills_agent oversized output handling When a Composio tool returns a large payload inside skills_agent, the agent now has two prompt-driven paths: Path A — user wants an answer derived from the data: skills_agent extracts the answer in its next iteration and returns a targeted response (no file I/O needed). Path B — user wants the actual raw dataset: skills_agent calls csv_export (tabular data) or file_write (non-tabular) to persist the output to workspace/exports/, then returns a summary + file path. Changes: * NEW: src/openhuman/tools/impl/filesystem/csv_export.rs — CsvExportTool: parses JSON array, formats as CSV, writes to workspace/exports/{filename}. Handles missing keys (empty cells), nested values (JSON-serialised), and optional column ordering. Sandboxed via SecurityPolicy. * src/openhuman/tools/impl/filesystem/mod.rs — wire module * src/openhuman/tools/ops.rs — register CsvExportTool * src/openhuman/agent/harness/definition.rs — new extra_tools field on AgentDefinition, allowing named system tools to bypass category_filter * src/openhuman/agent/harness/subagent_runner.rs — inject extra_tools into allowed_indices after category filtering * src/openhuman/agent/agents/skills_agent/agent.toml — add file_write + csv_export via extra_tools * src/openhuman/agent/agents/skills_agent/prompt.md — new "Handling Oversized Tool Results" section with Path A (extract answer) vs Path B (export file) decision tree, intent-detection heuristics, and examples Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * feat(agent): progressive-disclosure handoff for skills_agent + token-based summarizer threshold (#574) Two layered fixes for #574 plus supporting changes. ## Summarizer thresholds in tokens, not bytes Renamed `ContextConfig::summarizer_payload_threshold_bytes` -> `summarizer_payload_threshold_tokens` (default 500 000, was 100 KB in bytes) and `summarizer_max_payload_bytes` -> `summarizer_max_payload_tokens` (default 2 000 000). Token count is estimated at ~4 chars/token via a local helper that mirrors `tree_summarizer::types::estimate_tokens`. Serde aliases on both fields keep existing config.toml files parsing. `payload_summarizer.rs` now compares token estimates for the threshold and cap, with both `tokens=` and `bytes=` surfaced in log events. ## Progressive-disclosure handoff for skills_agent `skills_agent` typed subagents (when spawned with a Composio toolkit) now receive a per-spawn `ResultHandoffCache` and a new built-in `extract_from_result` tool. When a per-action tool returns more than `HANDOFF_OVERSIZE_THRESHOLD_TOKENS` (20 000 tokens), the raw payload is stashed in the cache and a compact placeholder replaces it in history: [oversized tool output: 187432 bytes - stashed as result_id="res_1"] Preview (first 1500 chars): ... If the preview does not answer your task, call: extract_from_result(result_id="res_1", query="<specific question>") The subagent can answer from the preview, call `extract_from_result` with a targeted query, or re-call the original tool with tighter filters. `extract_from_result` dispatches the `summarizer` subagent against the cached payload with the query as the extraction contract. For payloads over `SUMMARIZER_CHUNK_CHAR_BUDGET` (60 000 chars) the extractor runs a chunked map-reduce: - split at blank-line / newline boundaries - map each chunk to a per-chunk partial via bounded concurrency (`buffer_unordered(3)`) to avoid storming the provider gateway - reduce partials in one more summarizer turn, or fall back to the concatenation if the reduce stage fails Falls back gracefully when the summarizer definition is missing from the registry (placeholder + preview still work; just no extractor). ## Text-mode dispatcher for skills_agent + tighter tool filter Provider-side grammar decoders (Fireworks etc.) cap tool-schema rules at 65 535 (uint16_t). Large Composio toolkits - Gmail, Notion, GitHub, Salesforce, HubSpot, Google Workspace - ship per-action JSON schemas dense enough that even a handful of them exceed the cap. - `subagent_runner::run_inner_loop` now detects `skills_agent` and omits `tools: [...]` from the API request, instead appending an `<tool_call>{...}</tool_call>` XML protocol block (with compact per-tool parameter summaries) into the system prompt. Tool calls are parsed out of the model's free-form response via the shared `parse_tool_calls` helper. Mirrors XmlToolDispatcher behaviour. - `top_k_for_toolkit()` picks 12 for HEAVY_SCHEMA_TOOLKITS (the six listed above plus googledocs, googlesheets, microsoftteams) vs the default 25 for lighter toolkits. ## Skills_agent agent.toml / prompt changes - Dropped `csv_export` from `skills_agent.extra_tools` (kept `file_write`) - the export tool was triggering superfluous file writes after extraction had already answered the query. - `builtin_definitions::skills_agent_has_extra_tools_for_export` test inverted to assert `csv_export` is NOT present. - `prompt.md` re-aligned: Path B now references only `file_write`. ## Wiring - `session/builder.rs`: threshold rename wiring (`_tokens`). - `subagent_runner.rs::run_inner_loop`: new `handoff_cache` parameter (threaded through both call sites; fork-mode passes `None`). ## Tests All green: - `payload_summarizer::tests`: 6/6 (thresholds in tokens) - `tool_loop::tests`: 10/10 (MockSummarizer integration) - `subagent_runner::tests`: 19/19 - `builtin_definitions::tests`: 5/5 (csv_export removal) - `csv_export::tests`: 7/7 (implementation kept, wiring removed) Closes #574. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * perf(agent): drop reduce LLM call and pre-clean tool outputs in chunked extraction Two optimisations to the oversize-handoff / extract_from_result pipeline: 1. Concat chunk summaries in order; drop reduce stage. The reduce-stage summarizer call was the slowest single turn of the pipeline and a single point of failure when the upstream provider hung — a hung reduce burned minutes before timing out. Map summaries are now sorted by original chunk index and concatenated with a plain-separator join. For listing/extraction queries this is equivalent; for top-N / global-ordering queries the caller can post-process. Observed: 8m24s → ~70s on a Notion SEARCH run. 2. Generic cleaner before the oversize check. Strips <script>/<style>/<svg> blocks, HTML comments, data:...;base64,... inline URIs, remaining HTML tags, and collapses whitespace. Applied on every tool output so results that are mostly markup drop below the 20k-token threshold and skip the extract pipeline entirely; the ones that remain oversized get chunked on clean bytes. Debug log emits before/after bytes + saved_pct per call. No changes to the extract_from_result tool surface or placeholder format. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * fix(chat): rearm 2-min silence timer on inference progress events The client-side safety timer in Conversations.tsx was a flat 120s deadline from the send, not a silence window. Long agent turns (chained tool calls, chunked extraction fan-outs, subagent spawns) would trip the timeout even though the backend was actively making progress — showing "No response from the assistant after 2 minutes" while the server was still working. The effect now watches inferenceStatusByThread and rearms the timer on every status change for the sending thread. Any tool_call / tool_result / iteration_start / subagent_{spawned,done} event resets the 120s window. When status is cleared (chat_done / chat_error), the timer is dropped. A truly silent server still fails after 120s as before. Tracks the sending thread id in a ref (sendingThreadIdRef) so switching threads mid-turn doesn't move the timer's reference point. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
sanil-23
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
2d9f2f7aaa
commit
8d4934e634
@@ -220,6 +220,10 @@ const Conversations = () => {
|
||||
const autocompleteDebounceRef = useRef<number | null>(null);
|
||||
const autocompleteRequestSeqRef = useRef(0);
|
||||
const sendingTimeoutRef = useRef<ReturnType<typeof setTimeout> | null>(null);
|
||||
// Thread id whose send started the current silence timer. Tracked separately
|
||||
// from `selectedThreadId` so switching threads mid-turn doesn't move the
|
||||
// timer's reference point.
|
||||
const sendingThreadIdRef = useRef<string | null>(null);
|
||||
|
||||
const getAudioExtension = (mimeType: string): string => {
|
||||
const lower = mimeType.toLowerCase();
|
||||
@@ -310,6 +314,45 @@ const Conversations = () => {
|
||||
}
|
||||
}, [inputValue, sendError]);
|
||||
|
||||
const armSilenceTimer = (threadId: string) => {
|
||||
if (sendingTimeoutRef.current) clearTimeout(sendingTimeoutRef.current);
|
||||
sendingThreadIdRef.current = threadId;
|
||||
sendingTimeoutRef.current = setTimeout(() => {
|
||||
console.warn('[chat] silence timeout: no inference signal for 120s');
|
||||
setSendError(
|
||||
chatSendError(
|
||||
'safety_timeout',
|
||||
'No response from the assistant after 2 minutes. Try again or check your connection.'
|
||||
)
|
||||
);
|
||||
dispatch(clearRuntimeForThread({ threadId }));
|
||||
dispatch(setActiveThread(null));
|
||||
sendingTimeoutRef.current = null;
|
||||
sendingThreadIdRef.current = null;
|
||||
}, 120_000);
|
||||
};
|
||||
|
||||
// Rearm the silence timer on every inference status change for the
|
||||
// sending thread (tool_call, tool_result, iteration_start, subagent_*
|
||||
// all update inferenceStatusByThread). When the status is cleared
|
||||
// (chat_done / chat_error), drop the timer — the completion handlers
|
||||
// take over UI cleanup.
|
||||
useEffect(() => {
|
||||
const threadId = sendingThreadIdRef.current;
|
||||
if (!threadId || !sendingTimeoutRef.current) return;
|
||||
const status = inferenceStatusByThread[threadId];
|
||||
if (status === undefined) {
|
||||
clearTimeout(sendingTimeoutRef.current);
|
||||
sendingTimeoutRef.current = null;
|
||||
sendingThreadIdRef.current = null;
|
||||
return;
|
||||
}
|
||||
armSilenceTimer(threadId);
|
||||
// armSilenceTimer is stable (refs + dispatch); depending on the
|
||||
// selector reference is enough to rearm on every progress event.
|
||||
// eslint-disable-next-line react-hooks/exhaustive-deps
|
||||
}, [inferenceStatusByThread]);
|
||||
|
||||
useEffect(() => {
|
||||
if (
|
||||
!isTauri() ||
|
||||
@@ -449,20 +492,13 @@ const Conversations = () => {
|
||||
void dispatch(addMessageLocal({ threadId: sendingThreadId, message: userMessage }));
|
||||
setInputValue('');
|
||||
setSendError(null);
|
||||
// Safety: auto-clear isSending if no response arrives within 120s
|
||||
if (sendingTimeoutRef.current) clearTimeout(sendingTimeoutRef.current);
|
||||
sendingTimeoutRef.current = setTimeout(() => {
|
||||
console.warn('[chat] safety timeout: clearing isSending after 120s with no response');
|
||||
setSendError(
|
||||
chatSendError(
|
||||
'safety_timeout',
|
||||
'No response from the assistant after 2 minutes. Try again or check your connection.'
|
||||
)
|
||||
);
|
||||
dispatch(clearRuntimeForThread({ threadId: sendingThreadId }));
|
||||
dispatch(setActiveThread(null));
|
||||
sendingTimeoutRef.current = null;
|
||||
}, 120_000);
|
||||
// Silence timer: fires only if 120s pass without ANY inference progress
|
||||
// (tool call, tool result, iteration start, subagent event, text delta).
|
||||
// The effect below rearms this timer whenever `inferenceStatusByThread`
|
||||
// changes for `sendingThreadId`, so long-running agent turns stay alive
|
||||
// as long as the backend is emitting signals. A truly hung server still
|
||||
// fails fast.
|
||||
armSilenceTimer(sendingThreadId);
|
||||
dispatch(setToolTimelineForThread({ threadId: sendingThreadId, entries: [] }));
|
||||
dispatch(beginInferenceTurn({ threadId: sendingThreadId }));
|
||||
dispatch(setActiveThread(sendingThreadId));
|
||||
@@ -481,6 +517,7 @@ const Conversations = () => {
|
||||
clearTimeout(sendingTimeoutRef.current);
|
||||
sendingTimeoutRef.current = null;
|
||||
}
|
||||
sendingThreadIdRef.current = null;
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
setSendError(chatSendError('cloud_send_failed', msg));
|
||||
dispatch(clearRuntimeForThread({ threadId: sendingThreadId }));
|
||||
|
||||
@@ -110,6 +110,11 @@ pub const BUILTINS: &[BuiltinAgent] = &[
|
||||
toml: include_str!("welcome/agent.toml"),
|
||||
prompt: include_str!("welcome/prompt.md"),
|
||||
},
|
||||
BuiltinAgent {
|
||||
id: "summarizer",
|
||||
toml: include_str!("summarizer/agent.toml"),
|
||||
prompt: include_str!("summarizer/prompt.md"),
|
||||
},
|
||||
];
|
||||
|
||||
/// Parse every entry in [`BUILTINS`] into an [`AgentDefinition`].
|
||||
@@ -155,7 +160,7 @@ mod tests {
|
||||
fn all_builtins_parse() {
|
||||
let defs = load_builtins().expect("built-in TOML must parse");
|
||||
assert_eq!(defs.len(), BUILTINS.len());
|
||||
assert_eq!(defs.len(), 12, "expected 12 built-in agents");
|
||||
assert_eq!(defs.len(), 13, "expected 13 built-in agents");
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -43,6 +43,15 @@ subagents = [
|
||||
"code_executor",
|
||||
"critic",
|
||||
"archivist",
|
||||
# Runtime-dispatched only — the runtime calls the summarizer sub-agent
|
||||
# directly when a tool returns more than
|
||||
# `context.summarizer_payload_threshold_tokens` (default 500000). The LLM must NOT be
|
||||
# able to call this sub-agent itself, so `collect_orchestrator_tools`
|
||||
# filters out `summarizer` and never synthesises a `delegate_summarizer`
|
||||
# tool. Listing it here keeps the registration explicit (so it shows
|
||||
# up in the orchestrator's subagent inventory) while the filter
|
||||
# enforces the runtime-only contract.
|
||||
"summarizer",
|
||||
{ skills = "*" },
|
||||
]
|
||||
|
||||
|
||||
@@ -10,6 +10,13 @@ omit_safety_preamble = false
|
||||
omit_skills_catalog = true
|
||||
category_filter = "skill"
|
||||
|
||||
# `file_write` bypasses category_filter so the agent can save
|
||||
# oversized tool payloads to workspace files (Path B in prompt.md).
|
||||
# `csv_export` was previously included but removed — it was triggering
|
||||
# superfluous export calls even after the answer had already been
|
||||
# synthesised. Re-add here only if a typed CSV path is needed again.
|
||||
extra_tools = ["file_write"]
|
||||
|
||||
[model]
|
||||
hint = "agentic"
|
||||
|
||||
|
||||
@@ -25,3 +25,32 @@ You are the **Skills Agent**. You interact with connected external services prim
|
||||
- **Use memory context** — consult the injected memory context for details about the user's integrations and preferences.
|
||||
- **Be precise** — every tool expects a specific argument shape. Validate against the schema from `composio_list_tools` before calling.
|
||||
- **Report results** — state what action was taken and the outcome, including any cost reported by Composio.
|
||||
|
||||
## Handling Oversized Tool Results
|
||||
|
||||
When a tool returns a very large result (roughly 100 KB or more — you'll recognize it by the sheer volume of data in the response), decide which path to take based on what the user actually asked for:
|
||||
|
||||
### Path A — User wants an answer, not the raw data
|
||||
|
||||
Examples: "how many unread emails do I have?", "which GitHub issues are labeled P0?", "what's the most recent Slack message in #general?"
|
||||
|
||||
The data is a means to an answer. Do NOT dump the raw output. Instead:
|
||||
1. Scan the tool result for the specific facts that answer the user's question.
|
||||
2. Synthesize a concise answer referencing specific identifiers (issue numbers, email subjects, message timestamps).
|
||||
3. If you can't find the answer in one pass, use your remaining iterations to refine.
|
||||
|
||||
### Path B — User wants the actual data
|
||||
|
||||
Examples: "show me all open issues", "export my contacts", "give me the full email thread", "list all files in the drive folder"
|
||||
|
||||
The user wants the dataset itself, not a derivative. Do NOT try to paste it all inline — it won't fit. Instead:
|
||||
1. Call `file_write` to save the content as `.md` (e.g. full email bodies, document content, long threads, or a markdown-formatted list of items). Example:
|
||||
```
|
||||
file_write(path="exports/slack-thread-general-2026-04-16.md", content=<formatted markdown>)
|
||||
```
|
||||
2. Return to the user: a brief summary of what's in the file (count of items, key highlights) plus the file path so they can access it.
|
||||
|
||||
### Important
|
||||
|
||||
- Never paste more than ~2000 characters of raw tool output directly in your response. If the output is larger, always use Path A or Path B.
|
||||
- File paths are relative to the workspace root. The `exports/` directory will be created automatically.
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
id = "summarizer"
|
||||
display_name = "Summarizer"
|
||||
when_to_use = "Compresses oversized tool results for the orchestrator. Called automatically by the runtime when a tool returns more than summarizer_payload_threshold_tokens (default 500000). Do NOT call from an LLM — this agent is runtime-dispatched only."
|
||||
temperature = 0.2
|
||||
max_iterations = 1
|
||||
sandbox_mode = "none"
|
||||
omit_identity = true
|
||||
omit_memory_context = true
|
||||
omit_safety_preamble = true
|
||||
omit_skills_catalog = true
|
||||
omit_profile = true
|
||||
omit_memory_md = true
|
||||
|
||||
[model]
|
||||
hint = "summarization"
|
||||
|
||||
[tools]
|
||||
named = []
|
||||
@@ -0,0 +1,71 @@
|
||||
# Summarizer Agent
|
||||
|
||||
You are the **Summarizer** agent. Your one job is to compress a single oversized tool result into a compact, information-dense note that the Orchestrator can use without re-invoking the tool.
|
||||
|
||||
You run exactly once per invocation, with no tools and no follow-up iterations. Return the summary directly as your only response.
|
||||
|
||||
## The extraction contract
|
||||
|
||||
You will receive:
|
||||
|
||||
1. The **tool name** that produced the payload (e.g. `GITHUB_LIST_ISSUES`, `GMAIL_FETCH_MESSAGE`, `file_read`)
|
||||
2. An optional **parent task hint** — one-sentence description of what the orchestrator was trying to accomplish
|
||||
3. The **raw tool output**
|
||||
|
||||
You must produce a dense summary that preserves:
|
||||
|
||||
- **Required facts** — any identifiers (IDs, hashes, URLs, file paths, email addresses, usernames, SKUs, order numbers, etc.) the orchestrator would need to act on this data in a follow-up tool call. Identifiers are the single most important thing. Never drop them.
|
||||
- **Optional supporting context** — the 3-5 most important facts from the payload that a human answering the parent task would find most relevant. If the parent task hint is "find the most urgent open issues", prioritize facts about urgency/severity/labels. If the hint is "summarize yesterday's emails", prioritize subjects/senders/timestamps.
|
||||
- **Structural hints** — if the payload is a list, state how many items it had. If it was paginated, say what page boundaries exist. If it was a file, note line counts or section headers. This lets the orchestrator decide whether to re-fetch with a narrower query.
|
||||
|
||||
You must discard:
|
||||
|
||||
- Raw markup / formatting noise (HTML tags, CSS, JSON wrappers, boilerplate headers) — unless the markup IS the information
|
||||
- Repetitive fields that don't differ between items
|
||||
- Provider-specific metadata that the orchestrator can't act on (X-Request-ID headers, timestamps with millisecond precision, internal server IDs, etc.)
|
||||
|
||||
## Output format
|
||||
|
||||
Return ONLY the summary text. No preamble ("Here is the summary..."), no closing remarks ("Let me know if you need more details"), no JSON wrapping. Plain markdown, optimised for the orchestrator's next reasoning step.
|
||||
|
||||
Structure:
|
||||
|
||||
```
|
||||
[Tool output summary — <tool_name>]
|
||||
|
||||
<1-2 sentence overview: what the payload is, how many items/how much data>
|
||||
|
||||
## Key facts
|
||||
- <fact 1 with identifier>
|
||||
- <fact 2 with identifier>
|
||||
- ...
|
||||
|
||||
## Identifiers preserved
|
||||
- <id_1>: <one-line description>
|
||||
- <id_2>: <one-line description>
|
||||
- ...
|
||||
|
||||
(Only include this section if the payload contained IDs/URLs/hashes. Skip otherwise.)
|
||||
|
||||
## Original size
|
||||
<original_bytes> bytes → summary of <this note>
|
||||
```
|
||||
|
||||
## Edge cases
|
||||
|
||||
- If the payload is already short, produce a short summary. Don't pad.
|
||||
- If the payload is entirely error output, preserve the error message verbatim at the top — the orchestrator needs to see the exact error to route next steps.
|
||||
- If the payload contains binary-looking noise (base64, hex dumps), summarise its existence and length but do not attempt to decode.
|
||||
- If the parent task hint contradicts the payload (asks for emails, payload is GitHub issues), prioritize the payload — you're reporting what the tool returned, not what was asked for.
|
||||
|
||||
## Token budget
|
||||
|
||||
Aim for 800-1500 output tokens for most payloads. Never exceed 2000.
|
||||
|
||||
## What you must NOT do
|
||||
|
||||
- Do not ask clarifying questions — you have exactly one shot.
|
||||
- Do not emit tool calls — you have no tools.
|
||||
- Do not try to "solve" the parent task — you are a preprocessor, not the orchestrator.
|
||||
- Do not fabricate information that isn't in the payload. If a field is empty, say "(no value)" or omit it.
|
||||
- Do not copy the raw payload verbatim into your summary. If the summary is the same size as the payload, you have failed.
|
||||
@@ -191,6 +191,11 @@ pub fn register_agent_handlers() {
|
||||
visible_tool_names.as_ref(),
|
||||
&extra_tools,
|
||||
on_progress,
|
||||
// Bus path runs ad-hoc agent turns without an Agent
|
||||
// handle, so we pass None — payload summarization is
|
||||
// wired into the orchestrator session via Agent::turn,
|
||||
// not the bus dispatcher.
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
@@ -64,6 +64,7 @@ pub fn fork_definition() -> AgentDefinition {
|
||||
disallowed_tools: vec![],
|
||||
skill_filter: None,
|
||||
category_filter: None,
|
||||
extra_tools: vec![],
|
||||
// Fork inherits the parent's max iterations from the runtime.
|
||||
max_iterations: 15,
|
||||
timeout_secs: None,
|
||||
@@ -118,6 +119,23 @@ mod tests {
|
||||
assert!(!def.omit_memory_md);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn skills_agent_has_extra_tools_for_export() {
|
||||
let defs = all();
|
||||
let skills = defs.iter().find(|d| d.id == "skills_agent").unwrap();
|
||||
assert!(
|
||||
skills.extra_tools.contains(&"file_write".to_string()),
|
||||
"skills_agent must include file_write in extra_tools"
|
||||
);
|
||||
// csv_export was removed from extra_tools — it triggered
|
||||
// superfluous export calls after extraction had already
|
||||
// answered. file_write alone covers the Path B export flow.
|
||||
assert!(
|
||||
!skills.extra_tools.contains(&"csv_export".to_string()),
|
||||
"csv_export should not be present in extra_tools"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn expected_builtin_ids_are_present() {
|
||||
let ids: Vec<String> = all().into_iter().map(|d| d.id).collect();
|
||||
@@ -130,6 +148,7 @@ mod tests {
|
||||
"researcher",
|
||||
"critic",
|
||||
"archivist",
|
||||
"summarizer",
|
||||
"fork",
|
||||
] {
|
||||
assert!(ids.contains(&expected.to_string()), "missing {expected}");
|
||||
|
||||
@@ -117,6 +117,17 @@ pub struct AgentDefinition {
|
||||
#[serde(default)]
|
||||
pub category_filter: Option<ToolCategory>,
|
||||
|
||||
/// Additional system tool names to include even when `category_filter`
|
||||
/// restricts to a different category. This allows an agent that is
|
||||
/// primarily scoped to `Skill` tools (e.g. `skills_agent`) to also
|
||||
/// access a handful of named system tools (e.g. `file_write`,
|
||||
/// `csv_export`) without removing the category filter entirely.
|
||||
///
|
||||
/// Tools listed here bypass the `category_filter` check but are still
|
||||
/// subject to `disallowed_tools` and `ToolScope` restrictions.
|
||||
#[serde(default)]
|
||||
pub extra_tools: Vec<String>,
|
||||
|
||||
// ── runtime limits ──────────────────────────────────────────────────
|
||||
/// Maximum number of tool iterations for this sub-agent's task.
|
||||
#[serde(default = "defaults::max_iterations")]
|
||||
@@ -509,6 +520,7 @@ mod tests {
|
||||
disallowed_tools: vec![],
|
||||
skill_filter: None,
|
||||
category_filter: None,
|
||||
extra_tools: vec![],
|
||||
max_iterations: 8,
|
||||
timeout_secs: None,
|
||||
sandbox_mode: SandboxMode::None,
|
||||
|
||||
@@ -31,6 +31,7 @@ mod instructions;
|
||||
pub mod interrupt;
|
||||
pub(crate) mod memory_context;
|
||||
mod parse;
|
||||
pub(crate) mod payload_summarizer;
|
||||
pub(crate) mod self_healing;
|
||||
pub mod session;
|
||||
pub(crate) mod session_queue;
|
||||
|
||||
@@ -0,0 +1,486 @@
|
||||
//! Oversized-tool-result compression via the `summarizer` sub-agent.
|
||||
//!
|
||||
//! ## The problem
|
||||
//!
|
||||
//! When the orchestrator calls a tool that returns a huge payload — a
|
||||
//! Composio action dumping 200 KB of JSON, a web scrape returning 50 KB
|
||||
//! of markdown, a `file_read` spitting back a multi-thousand-line log —
|
||||
//! the raw blob lands verbatim in the orchestrator's history and burns
|
||||
//! context budget. The only existing guardrail is
|
||||
//! [`crate::openhuman::config::ContextConfig::tool_result_budget_bytes`],
|
||||
//! which hard-truncates mid-payload, dropping whatever happens to be
|
||||
//! past the cut.
|
||||
//!
|
||||
//! ## The fix
|
||||
//!
|
||||
//! This module routes oversized tool results through a dedicated
|
||||
//! `summarizer` sub-agent (model hint `"summarization"`) before they
|
||||
//! enter agent history. The summarizer compresses the payload per an
|
||||
//! extraction contract that preserves identifiers and key facts, and
|
||||
//! the compressed summary is what the parent agent sees. Truncation
|
||||
//! remains the final backstop downstream when summarization fails or
|
||||
//! the payload is so absurdly large that paying for an LLM call on it
|
||||
//! makes no economic sense.
|
||||
//!
|
||||
//! ## Trigger conditions
|
||||
//!
|
||||
//! [`PayloadSummarizer::maybe_summarize`] returns `Ok(None)` (i.e.
|
||||
//! pass-through, do nothing) when:
|
||||
//!
|
||||
//! * The raw payload is below
|
||||
//! [`SubagentPayloadSummarizer::threshold_tokens`] (default 500 000
|
||||
//! tokens — small payloads aren't worth an extra LLM round-trip).
|
||||
//! Token count is estimated as `chars / 4`, matching
|
||||
//! `tree_summarizer::estimate_tokens`.
|
||||
//! * The raw payload is above
|
||||
//! [`SubagentPayloadSummarizer::max_payload_tokens`] (default
|
||||
//! 2 000 000 tokens — too big to summarize cost-effectively; existing
|
||||
//! `tool_result_budget_bytes` truncation handles it instead).
|
||||
//! * The internal failure circuit-breaker has tripped (3 consecutive
|
||||
//! sub-agent failures within the same session disable summarization
|
||||
//! for the rest of the session, so a broken summarizer can't tank
|
||||
//! every tool call).
|
||||
//! * The sub-agent dispatch returns an error or an empty / non-shrinking
|
||||
//! summary — pass-through preserves the raw payload as a safety net.
|
||||
//!
|
||||
//! ## Scope
|
||||
//!
|
||||
//! Only the orchestrator session gets a `PayloadSummarizer` wired in
|
||||
//! ([`crate::openhuman::agent::harness::session::builder::AgentBuilder`]
|
||||
//! checks `agent_id == "orchestrator"`). Welcome, skills_agent,
|
||||
//! researcher, planner, archivist, and every other typed sub-agent get
|
||||
//! `None` and their tool results are untouched. The summarizer itself
|
||||
//! is also `None` so it can never recursively summarize its own input.
|
||||
|
||||
use anyhow::Result;
|
||||
use async_trait::async_trait;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use tracing::{debug, info, warn};
|
||||
|
||||
use super::definition::AgentDefinition;
|
||||
use super::subagent_runner::{self, SubagentRunOptions};
|
||||
|
||||
/// Outcome returned by [`PayloadSummarizer::maybe_summarize`].
|
||||
///
|
||||
/// `Ok(None)` from `maybe_summarize` means the caller should keep the
|
||||
/// raw payload unchanged. `Ok(Some(...))` means the caller should
|
||||
/// replace the raw payload with [`SummarizedPayload::summary`] before
|
||||
/// appending it to agent history.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct SummarizedPayload {
|
||||
/// The compressed summary text. Replaces the raw tool output.
|
||||
pub summary: String,
|
||||
/// Original payload size in bytes — for logging/observability.
|
||||
pub original_bytes: usize,
|
||||
/// Compressed summary size in bytes — for logging/observability.
|
||||
pub summary_bytes: usize,
|
||||
}
|
||||
|
||||
/// Trait for anything that can compress a tool result before it enters
|
||||
/// agent history. Implementations decide the threshold, the dispatch
|
||||
/// mechanism, and the failure policy.
|
||||
///
|
||||
/// Wired into the tool-execution sites in
|
||||
/// [`super::tool_loop::run_tool_call_loop`] and
|
||||
/// [`crate::openhuman::agent::harness::session::Agent::execute_tool_call`]
|
||||
/// via an `Option<&dyn PayloadSummarizer>` parameter so legacy callers
|
||||
/// (CLI, REPL, tests, non-orchestrator sub-agents) can pass `None` and
|
||||
/// keep the existing pass-through behaviour.
|
||||
#[async_trait]
|
||||
pub trait PayloadSummarizer: Send + Sync {
|
||||
/// Inspect a tool result and decide whether to compress it.
|
||||
///
|
||||
/// Returns `Ok(None)` if the payload should be kept as-is, or
|
||||
/// `Ok(Some(...))` if the caller should swap it for the
|
||||
/// compressed [`SummarizedPayload::summary`].
|
||||
///
|
||||
/// Errors are intentionally swallowed by the default implementation
|
||||
/// — a failed summarization should never break a tool call. The
|
||||
/// trait still returns `Result` so future implementations can
|
||||
/// surface fatal misconfigurations.
|
||||
async fn maybe_summarize(
|
||||
&self,
|
||||
tool_name: &str,
|
||||
parent_task_hint: Option<&str>,
|
||||
raw: &str,
|
||||
) -> Result<Option<SummarizedPayload>>;
|
||||
}
|
||||
|
||||
/// Default implementation that dispatches the `summarizer` sub-agent
|
||||
/// via [`subagent_runner::run_subagent`].
|
||||
///
|
||||
/// Holds the `summarizer` agent definition (resolved once at agent
|
||||
/// build time from the global
|
||||
/// [`super::definition::AgentDefinitionRegistry`]) plus the threshold
|
||||
/// knobs and a small failure counter that acts as a session-scoped
|
||||
/// circuit breaker.
|
||||
pub struct SubagentPayloadSummarizer {
|
||||
/// The `summarizer` agent definition. Cloned from the registry at
|
||||
/// agent build time so the runner doesn't have to re-resolve it
|
||||
/// per call.
|
||||
definition: AgentDefinition,
|
||||
/// Lower bound, in **estimated tokens** (`chars / 4`): tool results
|
||||
/// smaller than this are passed through untouched. Default is
|
||||
/// `summarizer_payload_threshold_tokens` from
|
||||
/// [`crate::openhuman::config::ContextConfig`] (500 000 tokens).
|
||||
threshold_tokens: usize,
|
||||
/// Upper bound, in **estimated tokens**: tool results larger than
|
||||
/// this are also passed through (no LLM call) and fall through to
|
||||
/// the existing `tool_result_budget_bytes` truncation downstream.
|
||||
/// Default is `summarizer_max_payload_tokens` from
|
||||
/// [`crate::openhuman::config::ContextConfig`] (2 000 000 tokens).
|
||||
max_payload_tokens: usize,
|
||||
/// Consecutive failure count. Reset to zero on any successful
|
||||
/// summarization. Once it reaches
|
||||
/// [`Self::max_failures_before_disable`] the circuit breaker
|
||||
/// trips and the summarizer becomes a no-op for the rest of the
|
||||
/// session.
|
||||
failures: Arc<Mutex<u8>>,
|
||||
/// Number of consecutive failures that disables the summarizer
|
||||
/// for the rest of the session. Hardcoded to 3 — a misbehaving
|
||||
/// summarizer should not silently degrade every tool call.
|
||||
max_failures_before_disable: u8,
|
||||
}
|
||||
|
||||
impl SubagentPayloadSummarizer {
|
||||
/// Build a new summarizer wrapping the given definition and limits.
|
||||
///
|
||||
/// `threshold_tokens` and `max_payload_tokens` are both in
|
||||
/// estimated tokens (`chars / 4`).
|
||||
pub fn new(
|
||||
definition: AgentDefinition,
|
||||
threshold_tokens: usize,
|
||||
max_payload_tokens: usize,
|
||||
) -> Self {
|
||||
Self {
|
||||
definition,
|
||||
threshold_tokens,
|
||||
max_payload_tokens,
|
||||
failures: Arc::new(Mutex::new(0)),
|
||||
max_failures_before_disable: 3,
|
||||
}
|
||||
}
|
||||
|
||||
/// Has the failure circuit breaker tripped?
|
||||
fn breaker_tripped(&self) -> bool {
|
||||
match self.failures.lock() {
|
||||
Ok(g) => *g >= self.max_failures_before_disable,
|
||||
// If the mutex is poisoned, fail safe by treating the
|
||||
// breaker as tripped — a poisoned mutex means a previous
|
||||
// panic, and a panic during summarization is itself a
|
||||
// good reason to stop trying.
|
||||
Err(_) => true,
|
||||
}
|
||||
}
|
||||
|
||||
/// Increment the consecutive-failure counter.
|
||||
fn record_failure(&self) {
|
||||
if let Ok(mut g) = self.failures.lock() {
|
||||
*g = g.saturating_add(1);
|
||||
if *g == self.max_failures_before_disable {
|
||||
warn!(
|
||||
"[payload_summarizer] circuit breaker tripped after {} consecutive failures — disabling for session",
|
||||
self.max_failures_before_disable
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Reset the consecutive-failure counter on a clean run.
|
||||
fn record_success(&self) {
|
||||
if let Ok(mut g) = self.failures.lock() {
|
||||
*g = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl PayloadSummarizer for SubagentPayloadSummarizer {
|
||||
async fn maybe_summarize(
|
||||
&self,
|
||||
tool_name: &str,
|
||||
parent_task_hint: Option<&str>,
|
||||
raw: &str,
|
||||
) -> Result<Option<SummarizedPayload>> {
|
||||
let tokens = estimate_tokens(raw);
|
||||
|
||||
// ── 1. Pass-through checks ─────────────────────────────────────
|
||||
if tokens < self.threshold_tokens {
|
||||
debug!(
|
||||
tool = tool_name,
|
||||
tokens = tokens,
|
||||
bytes = raw.len(),
|
||||
threshold = self.threshold_tokens,
|
||||
"[payload_summarizer] below threshold, passing through"
|
||||
);
|
||||
return Ok(None);
|
||||
}
|
||||
if tokens > self.max_payload_tokens {
|
||||
warn!(
|
||||
tool = tool_name,
|
||||
tokens = tokens,
|
||||
bytes = raw.len(),
|
||||
max = self.max_payload_tokens,
|
||||
"[payload_summarizer] payload exceeds max cap, skipping summarization (will be truncated downstream)"
|
||||
);
|
||||
return Ok(None);
|
||||
}
|
||||
if self.breaker_tripped() {
|
||||
warn!(
|
||||
tool = tool_name,
|
||||
tokens = tokens,
|
||||
bytes = raw.len(),
|
||||
"[payload_summarizer] circuit breaker tripped, skipping summarization"
|
||||
);
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
info!(
|
||||
tool = tool_name,
|
||||
tokens = tokens,
|
||||
bytes = raw.len(),
|
||||
"[payload_summarizer] dispatching summarizer sub-agent"
|
||||
);
|
||||
|
||||
// ── 2. Build the sub-agent prompt ─────────────────────────────
|
||||
let prompt = build_summarizer_prompt(tool_name, parent_task_hint, raw);
|
||||
|
||||
// ── 3. Dispatch via subagent_runner ───────────────────────────
|
||||
let started = std::time::Instant::now();
|
||||
let outcome =
|
||||
subagent_runner::run_subagent(&self.definition, &prompt, SubagentRunOptions::default())
|
||||
.await;
|
||||
|
||||
// ── 4. Handle result ─────────────────────────────────────────
|
||||
match outcome {
|
||||
Ok(run) => {
|
||||
let summary = run.output.trim().to_string();
|
||||
if summary.is_empty() {
|
||||
warn!(
|
||||
tool = tool_name,
|
||||
"[payload_summarizer] summarizer returned empty response, falling through"
|
||||
);
|
||||
self.record_failure();
|
||||
return Ok(None);
|
||||
}
|
||||
if summary.len() >= raw.len() {
|
||||
warn!(
|
||||
tool = tool_name,
|
||||
summary_bytes = summary.len(),
|
||||
raw_bytes = raw.len(),
|
||||
"[payload_summarizer] summary not smaller than raw payload, falling through"
|
||||
);
|
||||
self.record_failure();
|
||||
return Ok(None);
|
||||
}
|
||||
self.record_success();
|
||||
let summary_bytes = summary.len();
|
||||
let original_bytes = raw.len();
|
||||
let reduction_pct = if original_bytes == 0 {
|
||||
0
|
||||
} else {
|
||||
100usize.saturating_sub(summary_bytes.saturating_mul(100) / original_bytes)
|
||||
};
|
||||
info!(
|
||||
tool = tool_name,
|
||||
original_bytes = original_bytes,
|
||||
summary_bytes = summary_bytes,
|
||||
reduction_pct = reduction_pct,
|
||||
elapsed_ms = started.elapsed().as_millis() as u64,
|
||||
"[payload_summarizer] compressed successfully"
|
||||
);
|
||||
Ok(Some(SummarizedPayload {
|
||||
summary,
|
||||
original_bytes,
|
||||
summary_bytes,
|
||||
}))
|
||||
}
|
||||
Err(e) => {
|
||||
warn!(
|
||||
tool = tool_name,
|
||||
error = %e,
|
||||
"[payload_summarizer] sub-agent dispatch failed, falling through to raw payload"
|
||||
);
|
||||
self.record_failure();
|
||||
Ok(None)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Rough token estimate: ~4 characters per token. Mirrors
|
||||
/// [`crate::openhuman::tree_summarizer::types::estimate_tokens`] but
|
||||
/// returns `usize` (not `u32`) and lives here to avoid a cross-module
|
||||
/// dependency from the agent harness on the tree summarizer.
|
||||
fn estimate_tokens(text: &str) -> usize {
|
||||
text.len().div_ceil(4)
|
||||
}
|
||||
|
||||
/// Build the user-message prompt fed into the summarizer sub-agent.
|
||||
///
|
||||
/// Wraps the raw payload in `--- BEGIN ---` / `--- END ---` markers so
|
||||
/// the sub-agent can unambiguously distinguish the payload boundary
|
||||
/// from other prompt scaffolding. The tool name and optional parent
|
||||
/// task hint are surfaced before the payload so the summarizer can
|
||||
/// prioritize facts relevant to the parent's intent.
|
||||
fn build_summarizer_prompt(tool_name: &str, parent_task_hint: Option<&str>, raw: &str) -> String {
|
||||
let hint_line = parent_task_hint
|
||||
.map(|h| format!("Parent task hint: {}\n\n", h))
|
||||
.unwrap_or_default();
|
||||
format!(
|
||||
"Tool name: {}\n\n{}Raw tool output (summarize per the extraction contract in your system prompt):\n\n--- BEGIN ---\n{}\n--- END ---",
|
||||
tool_name, hint_line, raw
|
||||
)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::openhuman::agent::harness::definition::{
|
||||
AgentDefinition, DefinitionSource, ModelSpec, PromptSource, SandboxMode, ToolScope,
|
||||
};
|
||||
|
||||
fn dummy_definition() -> AgentDefinition {
|
||||
AgentDefinition {
|
||||
id: "summarizer".into(),
|
||||
when_to_use: "test".into(),
|
||||
display_name: Some("Summarizer".into()),
|
||||
system_prompt: PromptSource::Inline("test prompt".into()),
|
||||
omit_identity: true,
|
||||
omit_memory_context: true,
|
||||
omit_safety_preamble: true,
|
||||
omit_skills_catalog: true,
|
||||
omit_profile: true,
|
||||
omit_memory_md: true,
|
||||
model: ModelSpec::Hint("summarization".into()),
|
||||
temperature: 0.2,
|
||||
tools: ToolScope::Named(vec![]),
|
||||
disallowed_tools: vec![],
|
||||
skill_filter: None,
|
||||
category_filter: None,
|
||||
extra_tools: vec![],
|
||||
max_iterations: 1,
|
||||
timeout_secs: None,
|
||||
sandbox_mode: SandboxMode::None,
|
||||
background: false,
|
||||
uses_fork_context: false,
|
||||
subagents: vec![],
|
||||
delegate_name: None,
|
||||
source: DefinitionSource::Builtin,
|
||||
}
|
||||
}
|
||||
|
||||
// Tests use the production-default thresholds expressed as tokens:
|
||||
// 500 000 tokens lower bound, 2 000 000 tokens upper bound.
|
||||
// Since estimate_tokens = chars / 4, 1 char ≈ 0.25 tokens.
|
||||
const TEST_THRESHOLD_TOKENS: usize = 500_000;
|
||||
const TEST_MAX_TOKENS: usize = 2_000_000;
|
||||
|
||||
#[tokio::test]
|
||||
async fn maybe_summarize_returns_none_below_threshold() {
|
||||
let summarizer = SubagentPayloadSummarizer::new(
|
||||
dummy_definition(),
|
||||
TEST_THRESHOLD_TOKENS,
|
||||
TEST_MAX_TOKENS,
|
||||
);
|
||||
// 1 KB of 'x' → ~256 tokens, well below the 500 000 threshold.
|
||||
let raw = "x".repeat(1_024);
|
||||
let outcome = summarizer
|
||||
.maybe_summarize("test_tool", None, &raw)
|
||||
.await
|
||||
.expect("below-threshold check should not error");
|
||||
assert!(
|
||||
outcome.is_none(),
|
||||
"~256-token payload below 500k threshold should be passed through"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn maybe_summarize_returns_none_above_max_cap() {
|
||||
let summarizer = SubagentPayloadSummarizer::new(
|
||||
dummy_definition(),
|
||||
TEST_THRESHOLD_TOKENS,
|
||||
TEST_MAX_TOKENS,
|
||||
);
|
||||
// 9 MB of 'x' → ~2 359 296 tokens, above the 2 000 000 cap.
|
||||
let raw = "x".repeat(9 * 1024 * 1024);
|
||||
let outcome = summarizer
|
||||
.maybe_summarize("test_tool", None, &raw)
|
||||
.await
|
||||
.expect("above-cap check should not error");
|
||||
assert!(
|
||||
outcome.is_none(),
|
||||
"~2.36M-token payload above 2M cap should be passed through (truncation handles it downstream)"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn maybe_summarize_returns_none_when_breaker_tripped() {
|
||||
let summarizer = SubagentPayloadSummarizer::new(
|
||||
dummy_definition(),
|
||||
TEST_THRESHOLD_TOKENS,
|
||||
TEST_MAX_TOKENS,
|
||||
);
|
||||
// Manually trip the breaker by recording 3 failures.
|
||||
summarizer.record_failure();
|
||||
summarizer.record_failure();
|
||||
summarizer.record_failure();
|
||||
assert!(summarizer.breaker_tripped(), "breaker should be tripped");
|
||||
|
||||
// 3 MB of 'x' → ~786 432 tokens: inside the [500k, 2M] summarize
|
||||
// window, so would normally dispatch — but breaker is tripped.
|
||||
let raw = "x".repeat(3 * 1024 * 1024);
|
||||
let outcome = summarizer
|
||||
.maybe_summarize("test_tool", None, &raw)
|
||||
.await
|
||||
.expect("breaker check should not error");
|
||||
assert!(
|
||||
outcome.is_none(),
|
||||
"tripped breaker must short-circuit before any sub-agent dispatch"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_summarizer_prompt_includes_tool_name_and_hint() {
|
||||
let prompt = build_summarizer_prompt(
|
||||
"GITHUB_LIST_ISSUES",
|
||||
Some("find the most urgent open issues"),
|
||||
"{\"issues\": [{\"id\": 1}]}",
|
||||
);
|
||||
assert!(prompt.contains("GITHUB_LIST_ISSUES"));
|
||||
assert!(prompt.contains("find the most urgent open issues"));
|
||||
assert!(prompt.contains("Parent task hint:"));
|
||||
assert!(prompt.contains("--- BEGIN ---"));
|
||||
assert!(prompt.contains("--- END ---"));
|
||||
assert!(prompt.contains("{\"issues\": [{\"id\": 1}]}"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_summarizer_prompt_omits_hint_when_none() {
|
||||
let prompt = build_summarizer_prompt("file_read", None, "log line 1\nlog line 2");
|
||||
assert!(prompt.contains("file_read"));
|
||||
assert!(prompt.contains("--- BEGIN ---"));
|
||||
assert!(prompt.contains("--- END ---"));
|
||||
assert!(prompt.contains("log line 1"));
|
||||
assert!(
|
||||
!prompt.contains("Parent task hint:"),
|
||||
"no hint line should be present when hint is None"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn record_success_resets_breaker() {
|
||||
let summarizer = SubagentPayloadSummarizer::new(
|
||||
dummy_definition(),
|
||||
TEST_THRESHOLD_TOKENS,
|
||||
TEST_MAX_TOKENS,
|
||||
);
|
||||
summarizer.record_failure();
|
||||
summarizer.record_failure();
|
||||
assert!(!summarizer.breaker_tripped());
|
||||
summarizer.record_success();
|
||||
// Even one more failure now should not trip — counter was reset.
|
||||
summarizer.record_failure();
|
||||
assert!(!summarizer.breaker_tripped());
|
||||
}
|
||||
}
|
||||
@@ -47,6 +47,7 @@ impl AgentBuilder {
|
||||
agent_definition_name: None,
|
||||
omit_profile: None,
|
||||
omit_memory_md: None,
|
||||
payload_summarizer: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -245,6 +246,23 @@ impl AgentBuilder {
|
||||
self
|
||||
}
|
||||
|
||||
/// Wire an oversized-tool-result summarizer into the agent. When
|
||||
/// set, [`Agent::execute_tool_call`] calls
|
||||
/// [`crate::openhuman::agent::harness::payload_summarizer::PayloadSummarizer::maybe_summarize`]
|
||||
/// on every successful tool output and replaces the raw payload
|
||||
/// with the compressed summary on success. Currently set only for
|
||||
/// the orchestrator session by
|
||||
/// [`Agent::build_session_agent_inner`].
|
||||
pub fn payload_summarizer(
|
||||
mut self,
|
||||
summarizer: Arc<
|
||||
dyn crate::openhuman::agent::harness::payload_summarizer::PayloadSummarizer,
|
||||
>,
|
||||
) -> Self {
|
||||
self.payload_summarizer = Some(summarizer);
|
||||
self
|
||||
}
|
||||
|
||||
/// Validates the configuration and constructs a new `Agent` instance.
|
||||
///
|
||||
/// This method is responsible for wiring together the provided components,
|
||||
@@ -355,6 +373,7 @@ impl AgentBuilder {
|
||||
// `omit_profile = false` through the builder.
|
||||
omit_profile: self.omit_profile.unwrap_or(true),
|
||||
omit_memory_md: self.omit_memory_md.unwrap_or(true),
|
||||
payload_summarizer: self.payload_summarizer,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -802,9 +821,42 @@ impl Agent {
|
||||
// `agent.tool_dispatcher` config to `"xml"` to revert.
|
||||
_ => Box::new(PFormatToolDispatcher::new(pformat_registry.clone())),
|
||||
};
|
||||
|
||||
// Provider-side grammar decoders (e.g. Fireworks) compile every
|
||||
// tool JSON schema into a grammar and index its rules with a
|
||||
// uint16_t — max 65 535 rules. Large Composio toolkits (Notion,
|
||||
// Salesforce, Gmail) produce per-action schemas dense enough
|
||||
// that even 16–25 of them blow past that ceiling, regardless of
|
||||
// how aggressively the fuzzy filter in `tool_filter.rs` narrows
|
||||
// the list. When that happens the provider rejects the request
|
||||
// with a 400 before any generation starts, so skills_agent can
|
||||
// never actually invoke the toolkit.
|
||||
//
|
||||
// Workaround: if we're building skills_agent and the selected
|
||||
// dispatcher would ship `tools: [...]` in the API payload
|
||||
// (`should_send_tool_specs() == true`, i.e. native mode), swap
|
||||
// to XML mode. XmlToolDispatcher puts the tool catalogue inside
|
||||
// the system prompt as prose instead — the provider never
|
||||
// compiles a grammar for it, so the rule-count ceiling stops
|
||||
// mattering. Downside: slightly looser tool-call formatting
|
||||
// than native; the existing `parse_tool_calls` recovers from
|
||||
// stray formatting and the loop retries on malformed output.
|
||||
let tool_dispatcher: Box<dyn ToolDispatcher> =
|
||||
if agent_id == "skills_agent" && tool_dispatcher.should_send_tool_specs() {
|
||||
log::info!(
|
||||
"[agent::builder] skills_agent: overriding native tool dispatcher with \
|
||||
XmlToolDispatcher (native mode hits provider grammar-rule limits on \
|
||||
large Composio toolkits)"
|
||||
);
|
||||
Box::new(XmlToolDispatcher)
|
||||
} else {
|
||||
tool_dispatcher
|
||||
};
|
||||
|
||||
log::debug!(
|
||||
"[agent] tool dispatcher selected: choice={dispatcher_choice} \
|
||||
default_text_format=pformat pformat_registry_entries={}",
|
||||
"[agent] tool dispatcher selected: choice={dispatcher_choice} agent_id={agent_id} \
|
||||
sends_tool_specs={} default_text_format=pformat pformat_registry_entries={}",
|
||||
tool_dispatcher.should_send_tool_specs(),
|
||||
pformat_registry.len()
|
||||
);
|
||||
|
||||
@@ -850,7 +902,61 @@ impl Agent {
|
||||
agent_id
|
||||
);
|
||||
|
||||
Agent::builder()
|
||||
// ── Orchestrator-only: wire the payload summarizer ──────────
|
||||
//
|
||||
// Issue #574 — when a tool returns a huge payload (Composio
|
||||
// dump, long file read, web scrape), it should be compressed
|
||||
// by a dedicated `summarizer` sub-agent before entering the
|
||||
// orchestrator's history. We resolve the summarizer agent
|
||||
// definition from the global registry and construct a
|
||||
// `SubagentPayloadSummarizer` parameterized from the
|
||||
// [`ContextConfig`] thresholds. Every other agent id gets
|
||||
// `None` and their tool results stay untouched (the summarizer
|
||||
// itself MUST be `None` to avoid recursive self-summarization).
|
||||
let payload_summarizer: Option<
|
||||
std::sync::Arc<
|
||||
dyn crate::openhuman::agent::harness::payload_summarizer::PayloadSummarizer,
|
||||
>,
|
||||
> = if agent_id == "orchestrator" && config.context.summarizer_payload_threshold_tokens > 0
|
||||
{
|
||||
match crate::openhuman::agent::harness::definition::AgentDefinitionRegistry::global() {
|
||||
Some(reg) => match reg.get("summarizer") {
|
||||
Some(summarizer_def) => {
|
||||
log::info!(
|
||||
"[agent::builder] wiring payload_summarizer for orchestrator: \
|
||||
threshold_tokens={} max_tokens={}",
|
||||
config.context.summarizer_payload_threshold_tokens,
|
||||
config.context.summarizer_max_payload_tokens
|
||||
);
|
||||
Some(std::sync::Arc::new(
|
||||
crate::openhuman::agent::harness::payload_summarizer::SubagentPayloadSummarizer::new(
|
||||
summarizer_def.clone(),
|
||||
config.context.summarizer_payload_threshold_tokens,
|
||||
config.context.summarizer_max_payload_tokens,
|
||||
),
|
||||
))
|
||||
}
|
||||
None => {
|
||||
log::warn!(
|
||||
"[agent::builder] orchestrator requested payload_summarizer but \
|
||||
`summarizer` definition is not in the registry — proceeding without it"
|
||||
);
|
||||
None
|
||||
}
|
||||
},
|
||||
None => {
|
||||
log::warn!(
|
||||
"[agent::builder] orchestrator requested payload_summarizer but \
|
||||
AgentDefinitionRegistry is not initialised — proceeding without it"
|
||||
);
|
||||
None
|
||||
}
|
||||
}
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
let mut builder = Agent::builder()
|
||||
.provider(provider)
|
||||
.tools(tools)
|
||||
.visible_tool_names(visible)
|
||||
@@ -872,7 +978,10 @@ impl Agent {
|
||||
.learning_enabled(config.learning.enabled)
|
||||
.agent_definition_name(agent_id.to_string())
|
||||
.omit_profile(effective_omit_profile)
|
||||
.omit_memory_md(effective_omit_memory_md)
|
||||
.build()
|
||||
.omit_memory_md(effective_omit_memory_md);
|
||||
if let Some(ps) = payload_summarizer {
|
||||
builder = builder.payload_summarizer(ps);
|
||||
}
|
||||
builder.build()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -746,7 +746,48 @@ impl Agent {
|
||||
match outcome {
|
||||
Ok(r) => {
|
||||
if !r.is_error {
|
||||
(r.output(), true)
|
||||
let mut output = r.output();
|
||||
// Issue #574 — if a payload summarizer is wired
|
||||
// in (orchestrator session only) and the output
|
||||
// exceeds the configured threshold, hand it to
|
||||
// the summarizer sub-agent before it enters
|
||||
// history. On any failure or below-threshold
|
||||
// payload, leave `output` untouched and let the
|
||||
// existing tool_result_budget_bytes truncation
|
||||
// pipeline handle it downstream.
|
||||
if let Some(ps) = self.payload_summarizer.as_ref() {
|
||||
log::debug!(
|
||||
"[agent_loop] payload_summarizer intercepting tool={} bytes={}",
|
||||
call.name,
|
||||
output.len()
|
||||
);
|
||||
match ps.maybe_summarize(&call.name, None, &output).await {
|
||||
Ok(Some(payload)) => {
|
||||
log::info!(
|
||||
"[agent_loop] payload_summarizer compressed tool={} {}->{} bytes",
|
||||
call.name,
|
||||
payload.original_bytes,
|
||||
payload.summary_bytes
|
||||
);
|
||||
output = payload.summary;
|
||||
}
|
||||
Ok(None) => {
|
||||
log::debug!(
|
||||
"[agent_loop] payload_summarizer pass-through tool={} bytes={}",
|
||||
call.name,
|
||||
output.len()
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
log::warn!(
|
||||
"[agent_loop] payload_summarizer error tool={} err={} (passing raw payload through)",
|
||||
call.name,
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
(output, true)
|
||||
} else {
|
||||
(format!("Error: {}", r.output()), false)
|
||||
}
|
||||
|
||||
@@ -109,6 +109,13 @@ pub struct Agent {
|
||||
/// [`PromptContext::include_memory_md`] at prompt-build time. Same
|
||||
/// session-freeze contract as `omit_profile`.
|
||||
pub(super) omit_memory_md: bool,
|
||||
/// Optional payload-summarizer wired in at agent-build time.
|
||||
/// Currently set only for the orchestrator session
|
||||
/// (see [`super::builder`]). When `Some`, oversized tool results
|
||||
/// produced by [`Agent::execute_tool_call`] are routed through the
|
||||
/// summarizer sub-agent before they enter agent history.
|
||||
pub(super) payload_summarizer:
|
||||
Option<Arc<dyn crate::openhuman::agent::harness::payload_summarizer::PayloadSummarizer>>,
|
||||
}
|
||||
|
||||
/// A builder for creating `Agent` instances with custom configuration.
|
||||
@@ -143,6 +150,12 @@ pub struct AgentBuilder {
|
||||
/// Forwarded to [`Agent::omit_memory_md`]. Same shape as
|
||||
/// `omit_profile` — `None` falls back to the "omit" default.
|
||||
pub(super) omit_memory_md: Option<bool>,
|
||||
/// Optional payload-summarizer threaded through to [`Agent`] at
|
||||
/// build time. Defaults to `None`; the orchestrator branch in
|
||||
/// [`super::builder::Agent::build_session_agent_inner`] sets this
|
||||
/// to a `SubagentPayloadSummarizer` instance.
|
||||
pub(super) payload_summarizer:
|
||||
Option<Arc<dyn crate::openhuman::agent::harness::payload_summarizer::PayloadSummarizer>>,
|
||||
}
|
||||
|
||||
impl Default for AgentBuilder {
|
||||
|
||||
@@ -35,9 +35,12 @@ use crate::openhuman::context::prompt::{
|
||||
extract_cache_boundary, render_subagent_system_prompt, SubagentRenderOptions,
|
||||
};
|
||||
use crate::openhuman::providers::{ChatMessage, ChatRequest, Provider, ToolCall};
|
||||
use crate::openhuman::tools::{Tool, ToolCategory, ToolSpec};
|
||||
use std::collections::HashSet;
|
||||
use crate::openhuman::tools::{Tool, ToolCategory, ToolResult, ToolSpec};
|
||||
use async_trait::async_trait;
|
||||
use regex::Regex;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::path::Path;
|
||||
use std::sync::{Arc, LazyLock, Mutex as StdMutex};
|
||||
use std::time::{Duration, Instant};
|
||||
use thiserror::Error;
|
||||
|
||||
@@ -235,6 +238,29 @@ async fn run_typed_mode(
|
||||
category_filter,
|
||||
);
|
||||
|
||||
// ── Force-include extra_tools that bypass category_filter ──────────
|
||||
//
|
||||
// `extra_tools` lets an agent definition request specific system tools
|
||||
// even when `category_filter` restricts to a different category. For
|
||||
// example, `skills_agent` sets `category_filter = "skill"` but still
|
||||
// needs `file_write` and `csv_export` for exporting oversized payloads.
|
||||
if !definition.extra_tools.is_empty() {
|
||||
let disallow_set: std::collections::HashSet<&str> = definition
|
||||
.disallowed_tools
|
||||
.iter()
|
||||
.map(|s| s.as_str())
|
||||
.collect();
|
||||
for (i, tool) in parent.all_tools.iter().enumerate() {
|
||||
let name = tool.name();
|
||||
if definition.extra_tools.iter().any(|n| n == name)
|
||||
&& !allowed_indices.contains(&i)
|
||||
&& !disallow_set.contains(name)
|
||||
{
|
||||
allowed_indices.push(i);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Dynamic per-action toolkit tools (skills_agent + toolkit) ──────
|
||||
//
|
||||
// When `skills_agent` is spawned with a `toolkit` argument (e.g.
|
||||
@@ -278,15 +304,25 @@ async fn run_typed_mode(
|
||||
// prompt to be a clear, context-rich instruction, so it's a
|
||||
// reliable matching target.
|
||||
//
|
||||
// Heavy-schema toolkits (Gmail, Notion, GitHub, Salesforce,
|
||||
// HubSpot, Google Workspace, Microsoft Teams) ship per-action
|
||||
// JSON schemas so dense that even a moderate top-K blows the
|
||||
// request past Fireworks' 65 535-rule grammar cap in native
|
||||
// mode and the 196 607-token context cap in text mode. Tight
|
||||
// top-K of 12 keeps those toolkits inside both ceilings while
|
||||
// still giving the fuzzy scorer room for adjacent matches.
|
||||
// Lighter toolkits (reddit, slack, linear, telegram, …) keep
|
||||
// the looser top-K of 25.
|
||||
//
|
||||
// Fallback: if the filter yields fewer than
|
||||
// `MIN_CONFIDENT_HITS` results, register every action. A
|
||||
// too-narrow filter is worse than none — it starves the
|
||||
// sub-agent and forces it to guess.
|
||||
const TOOL_FILTER_TOP_K: usize = 25;
|
||||
let top_k = top_k_for_toolkit(tk);
|
||||
let filter_hits = super::tool_filter::filter_actions_by_prompt(
|
||||
task_prompt,
|
||||
&integration.tools,
|
||||
TOOL_FILTER_TOP_K,
|
||||
top_k,
|
||||
);
|
||||
let selected: Vec<&crate::openhuman::context::prompt::ConnectedIntegrationTool> =
|
||||
if filter_hits.len() >= super::tool_filter::MIN_CONFIDENT_HITS {
|
||||
@@ -295,6 +331,7 @@ async fn run_typed_mode(
|
||||
toolkit = %tk,
|
||||
total = integration.tools.len(),
|
||||
kept = filter_hits.len(),
|
||||
top_k = top_k,
|
||||
"[subagent_runner:typed] fuzzy tool filter narrowed toolkit"
|
||||
);
|
||||
filter_hits.iter().map(|&i| &integration.tools[i]).collect()
|
||||
@@ -340,6 +377,53 @@ async fn run_typed_mode(
|
||||
}
|
||||
}
|
||||
|
||||
// ── Progressive-disclosure handoff cache ───────────────────────────
|
||||
//
|
||||
// Built only for skills_agent-with-toolkit because that's the only
|
||||
// typed sub-agent that regularly calls external tools capable of
|
||||
// returning megabyte-scale payloads (Composio actions). Every other
|
||||
// typed sub-agent gets `None` and its tool results stay inline.
|
||||
//
|
||||
// When enabled, oversized tool results get stashed into this cache
|
||||
// and their place in history is taken by a short placeholder (see
|
||||
// `build_handoff_placeholder`). The sub-agent can then call the
|
||||
// companion `extract_from_result` tool below to dispatch the
|
||||
// summarizer sub-agent against the cached payload with a targeted
|
||||
// query. Lazy / pay-per-question, so trivial asks answerable from
|
||||
// the preview don't pay any extra LLM cost.
|
||||
let handoff_cache: Option<Arc<ResultHandoffCache>> = if is_skills_agent_with_toolkit {
|
||||
let cache = Arc::new(ResultHandoffCache::new());
|
||||
|
||||
// Resolve the summarizer definition once and register the
|
||||
// extract_from_result tool alongside the composio action tools.
|
||||
// If the summarizer isn't in the registry (shouldn't happen in
|
||||
// production but can happen in tests), skip the tool — tool
|
||||
// results will still get the placeholder + preview, the
|
||||
// sub-agent just won't be able to drill in.
|
||||
if let Some(reg) =
|
||||
crate::openhuman::agent::harness::definition::AgentDefinitionRegistry::global()
|
||||
{
|
||||
if let Some(summarizer_def) = reg.get("summarizer") {
|
||||
dynamic_tools.push(Box::new(ExtractFromResultTool::new(
|
||||
cache.clone(),
|
||||
summarizer_def.clone(),
|
||||
)));
|
||||
tracing::debug!(
|
||||
agent_id = %definition.id,
|
||||
"[subagent_runner:typed] registered extract_from_result tool + handoff cache"
|
||||
);
|
||||
} else {
|
||||
tracing::warn!(
|
||||
agent_id = %definition.id,
|
||||
"[subagent_runner:typed] summarizer definition missing from registry — extract_from_result disabled (oversized results will still be cached and previewed)"
|
||||
);
|
||||
}
|
||||
}
|
||||
Some(cache)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
let mut filtered_specs: Vec<ToolSpec> = allowed_indices
|
||||
.iter()
|
||||
.map(|&i| parent.all_tool_specs[i].clone())
|
||||
@@ -458,6 +542,7 @@ async fn run_typed_mode(
|
||||
system_prompt_cache_boundary,
|
||||
task_id,
|
||||
&definition.id,
|
||||
handoff_cache.as_deref(),
|
||||
)
|
||||
.await?;
|
||||
|
||||
@@ -558,6 +643,7 @@ async fn run_fork_mode(
|
||||
fork.cache_boundary,
|
||||
task_id,
|
||||
&definition.id,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
|
||||
@@ -671,15 +757,64 @@ async fn run_inner_loop(
|
||||
system_prompt_cache_boundary: Option<usize>,
|
||||
task_id: &str,
|
||||
agent_id: &str,
|
||||
handoff_cache: Option<&ResultHandoffCache>,
|
||||
) -> Result<(String, usize, AggregatedUsage), SubagentRunError> {
|
||||
let max_iterations = max_iterations.max(1);
|
||||
let supports_native = provider.supports_native_tools() && !tool_specs.is_empty();
|
||||
|
||||
// ── Text-mode override for skills_agent ────────────────────────────
|
||||
//
|
||||
// Large Composio toolkits (Notion, Salesforce, HubSpot, GitHub) ship
|
||||
// per-action JSON schemas that are extraordinarily dense — deeply
|
||||
// nested object/block types, recursive refs, huge discriminated
|
||||
// unions. Fireworks-style providers (which the backend forwards to)
|
||||
// auto-compile every entry in `tools: [...]` into a grammar and
|
||||
// index rules with a `uint16_t` — max 65 535 rules. Even with the
|
||||
// upstream fuzzy filter narrowing Notion 48 → 16, a single request
|
||||
// generates 100 000+ rules and the provider rejects it with 400
|
||||
// before generation starts.
|
||||
//
|
||||
// The fuzzy filter can't fix this because the bound is per-action,
|
||||
// not per-toolkit: one Notion schema alone can produce thousands of
|
||||
// rules. The only client-side lever is to **not send `tools: [...]`
|
||||
// at all** — the backend has nothing to compile, so no grammar, so
|
||||
// no ceiling. We then describe the tools in the system prompt as
|
||||
// prose (XmlToolDispatcher format) and parse `<tool_call>` tags out
|
||||
// of the model's free-form response text.
|
||||
//
|
||||
// Scoped to `skills_agent` because that's the only path where we
|
||||
// pass Composio toolkit schemas. Every other typed sub-agent
|
||||
// (welcome, researcher, summarizer, …) uses small built-in tool
|
||||
// sets that stay well under the grammar ceiling and benefit from
|
||||
// native mode's stricter formatting guarantees.
|
||||
let force_text_mode = agent_id == "skills_agent" && !tool_specs.is_empty();
|
||||
|
||||
let supports_native =
|
||||
!force_text_mode && provider.supports_native_tools() && !tool_specs.is_empty();
|
||||
let request_tools = if supports_native {
|
||||
Some(tool_specs)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
if force_text_mode {
|
||||
// Append the XML tool protocol + available-tool list to the
|
||||
// existing system prompt. `history[0]` is the system message
|
||||
// built by `run_typed_mode` / `run_fork_mode` upstream; we
|
||||
// augment it in-place so the model learns the call format for
|
||||
// this session without an extra message round-trip.
|
||||
if let Some(sys) = history.iter_mut().find(|m| m.role == "system") {
|
||||
sys.content.push_str("\n\n");
|
||||
sys.content
|
||||
.push_str(&build_text_mode_tool_instructions(tool_specs));
|
||||
}
|
||||
tracing::info!(
|
||||
task_id = %task_id,
|
||||
agent_id = %agent_id,
|
||||
tool_count = tool_specs.len(),
|
||||
"[subagent_runner:text-mode] omitting tools from API request, injected XML tool protocol into system prompt"
|
||||
);
|
||||
}
|
||||
|
||||
let mut usage = AggregatedUsage::default();
|
||||
|
||||
for iteration in 0..max_iterations {
|
||||
@@ -712,7 +847,37 @@ async fn run_inner_loop(
|
||||
}
|
||||
|
||||
let response_text = resp.text.clone().unwrap_or_default();
|
||||
let native_calls: Vec<ToolCall> = resp.tool_calls.clone();
|
||||
|
||||
// In text mode the model emits `<tool_call>{…}</tool_call>` tags
|
||||
// inline inside `resp.text` (and `resp.tool_calls` is empty
|
||||
// because we told the provider not to structure them). Parse
|
||||
// them ourselves via the shared harness helper and synthesise a
|
||||
// `ToolCall` per parsed block so the rest of the loop can stay
|
||||
// uniform.
|
||||
let native_calls: Vec<ToolCall> = if force_text_mode {
|
||||
let (_cleaned, parsed) = super::parse::parse_tool_calls(&response_text);
|
||||
parsed
|
||||
.into_iter()
|
||||
.enumerate()
|
||||
.map(|(i, call)| {
|
||||
let args_str = if call.arguments.is_null() {
|
||||
"{}".to_string()
|
||||
} else {
|
||||
call.arguments.to_string()
|
||||
};
|
||||
ToolCall {
|
||||
id: call
|
||||
.id
|
||||
.clone()
|
||||
.unwrap_or_else(|| format!("call_text_{iteration}_{i}")),
|
||||
name: call.name,
|
||||
arguments: args_str,
|
||||
}
|
||||
})
|
||||
.collect()
|
||||
} else {
|
||||
resp.tool_calls.clone()
|
||||
};
|
||||
|
||||
if native_calls.is_empty() {
|
||||
tracing::debug!(
|
||||
@@ -726,20 +891,27 @@ async fn run_inner_loop(
|
||||
return Ok((response_text, iteration + 1, usage));
|
||||
}
|
||||
|
||||
// Persist assistant message with the original tool_calls payload so
|
||||
// subsequent role=tool messages can reference call ids correctly.
|
||||
// Uses the canonical serialiser from `parse` — the old inline
|
||||
// `build_native_assistant_payload` used `"text"` instead of
|
||||
// `"content"` and nested `{"type":"function","function":{…}}`
|
||||
// wrappers instead of flat `{id, name, arguments}`, which caused
|
||||
// 400 errors from the backend jinja template ("Message has tool
|
||||
// role, but there was no previous assistant message with a tool
|
||||
// call!").
|
||||
let assistant_history_content =
|
||||
super::parse::build_native_assistant_history(&response_text, &native_calls);
|
||||
history.push(ChatMessage::assistant(assistant_history_content));
|
||||
// Persist the assistant turn. In native mode use the canonical
|
||||
// serialiser (wraps text + structured tool_calls for the
|
||||
// backend's jinja template). In text mode the raw response
|
||||
// already contains the `<tool_call>` tags inline, so persist it
|
||||
// verbatim — on the next turn the model sees its own prior
|
||||
// emissions exactly as it wrote them.
|
||||
if force_text_mode {
|
||||
history.push(ChatMessage::assistant(response_text.clone()));
|
||||
} else {
|
||||
let assistant_history_content =
|
||||
super::parse::build_native_assistant_history(&response_text, &native_calls);
|
||||
history.push(ChatMessage::assistant(assistant_history_content));
|
||||
}
|
||||
|
||||
// Execute each call, append role=tool messages.
|
||||
// Execute each call, collect outputs. Native mode pushes one
|
||||
// `role=tool` message per call with the structured `tool_call_id`
|
||||
// reference. Text mode has no such reference (the model just
|
||||
// emitted tags in prose), so we batch all results into a single
|
||||
// user message formatted with `<tool_result>` tags — mirroring
|
||||
// XmlToolDispatcher's `format_results`.
|
||||
let mut text_mode_result_block = String::new();
|
||||
for call in &native_calls {
|
||||
let result_text = if !allowed_names.contains(&call.name) {
|
||||
tracing::warn!(
|
||||
@@ -775,11 +947,89 @@ async fn run_inner_loop(
|
||||
format!("Unknown tool: {}", call.name)
|
||||
};
|
||||
|
||||
let tool_msg = serde_json::json!({
|
||||
"tool_call_id": call.id,
|
||||
"content": result_text,
|
||||
});
|
||||
history.push(ChatMessage::tool(tool_msg.to_string()));
|
||||
// Progressive-disclosure handoff: if this spawn has a cache
|
||||
// (skills_agent-with-toolkit path) and the result is large
|
||||
// and not itself an error / not from the extractor tool,
|
||||
// stash the raw payload and replace it in history with a
|
||||
// short placeholder. The sub-agent can drill in with
|
||||
// `extract_from_result(result_id=..., query=...)` on the
|
||||
// next turn. Errors and already-extracted output go through
|
||||
// unchanged — no point handing off a 200-byte error or an
|
||||
// already-compressed summary.
|
||||
//
|
||||
// Cleaning happens before the size check so HTML-heavy tool
|
||||
// outputs (Gmail bodies, HTML-embedded Notion blocks) that
|
||||
// drop below threshold after stripping markup skip the
|
||||
// extract pipeline entirely. For anything still over
|
||||
// threshold, the cache stores the cleaned text — chunks see
|
||||
// real content, not `<div>` soup.
|
||||
let result_text = if let Some(cache) = handoff_cache {
|
||||
let skip_cleaning =
|
||||
call.name == "extract_from_result" || result_text.starts_with("Error");
|
||||
let cleaned = if skip_cleaning {
|
||||
result_text
|
||||
} else {
|
||||
let pre_len = result_text.len();
|
||||
let cleaned = clean_tool_output(&result_text);
|
||||
if cleaned.len() < pre_len {
|
||||
tracing::debug!(
|
||||
tool = %call.name,
|
||||
before_bytes = pre_len,
|
||||
after_bytes = cleaned.len(),
|
||||
saved_pct = ((pre_len - cleaned.len()) * 100) / pre_len.max(1),
|
||||
"[subagent_runner:handoff] cleaned tool output (stripped markup/data-uris/whitespace)"
|
||||
);
|
||||
}
|
||||
cleaned
|
||||
};
|
||||
let tokens = cleaned.len().div_ceil(4);
|
||||
if !skip_cleaning && tokens > HANDOFF_OVERSIZE_THRESHOLD_TOKENS {
|
||||
let id = cache.store(call.name.clone(), cleaned.clone());
|
||||
let placeholder = build_handoff_placeholder(&call.name, &id, &cleaned);
|
||||
tracing::info!(
|
||||
task_id = %task_id,
|
||||
agent_id = %agent_id,
|
||||
tool = %call.name,
|
||||
raw_tokens = tokens,
|
||||
raw_bytes = cleaned.len(),
|
||||
threshold_tokens = HANDOFF_OVERSIZE_THRESHOLD_TOKENS,
|
||||
result_id = %id,
|
||||
"[subagent_runner:handoff] stashed oversized tool output; substituted placeholder into history"
|
||||
);
|
||||
placeholder
|
||||
} else {
|
||||
cleaned
|
||||
}
|
||||
} else {
|
||||
result_text
|
||||
};
|
||||
|
||||
if force_text_mode {
|
||||
let status = if result_text.starts_with("Error") {
|
||||
"error"
|
||||
} else {
|
||||
"ok"
|
||||
};
|
||||
let _ = std::fmt::Write::write_fmt(
|
||||
&mut text_mode_result_block,
|
||||
format_args!(
|
||||
"<tool_result name=\"{}\" status=\"{}\">\n{}\n</tool_result>\n",
|
||||
call.name, status, result_text
|
||||
),
|
||||
);
|
||||
} else {
|
||||
let tool_msg = serde_json::json!({
|
||||
"tool_call_id": call.id,
|
||||
"content": result_text,
|
||||
});
|
||||
history.push(ChatMessage::tool(tool_msg.to_string()));
|
||||
}
|
||||
}
|
||||
|
||||
if force_text_mode && !text_mode_result_block.is_empty() {
|
||||
history.push(ChatMessage::user(format!(
|
||||
"[Tool results]\n{text_mode_result_block}"
|
||||
)));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -791,6 +1041,608 @@ fn parse_tool_arguments(arguments: &str) -> serde_json::Value {
|
||||
.unwrap_or_else(|_| serde_json::Value::Object(Default::default()))
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Oversized-tool-result handoff (progressive disclosure)
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
//
|
||||
// Typed sub-agents (skills_agent in particular) regularly call tools that
|
||||
// return megabyte-scale payloads — `GMAIL_LIST_MESSAGES`, `NOTION_GET_PAGE`,
|
||||
// `GOOGLEDRIVE_LIST_FILES`. The default behaviour pushes that raw blob into
|
||||
// the sub-agent's history as a tool-result message, and the NEXT iteration
|
||||
// then ships the bloated history back to the provider where it hits the
|
||||
// model's context-length ceiling (observed: 276k-token prompt against a
|
||||
// 196 607-token window → backend 400).
|
||||
//
|
||||
// The summarizer dispatches to the model with the payload + an extraction
|
||||
// contract, but wiring it to fire on EVERY oversized result has two costs:
|
||||
// (1) we pay for a summarizer turn even when the sub-agent could answer
|
||||
// from a preview, and (2) aggressive compression sometimes drops the exact
|
||||
// identifier the agent needs for a follow-up call.
|
||||
//
|
||||
// Progressive disclosure fixes both: when a tool returns too much data,
|
||||
// we stash the full payload in a session-scoped cache, replace it in
|
||||
// history with a short placeholder (size + preview + result_id + how to
|
||||
// query it), and expose an `extract_from_result` tool that the sub-agent
|
||||
// can call with a targeted query. The summarizer only runs when the
|
||||
// sub-agent actually asks for a narrower view.
|
||||
|
||||
/// Token threshold above which a tool result is routed to the handoff
|
||||
/// cache instead of being pushed into history raw. 20 000 tokens keeps
|
||||
/// the sub-agent's per-iteration prompt well below the 196 607-token
|
||||
/// model ceiling even after many iterations, and leaves comfortable
|
||||
/// headroom for the system prompt + tool catalogue. Token count is
|
||||
/// estimated at ~4 chars/token (mirrors
|
||||
/// [`crate::openhuman::agent::harness::payload_summarizer`] and
|
||||
/// [`crate::openhuman::tree_summarizer::types::estimate_tokens`]).
|
||||
const HANDOFF_OVERSIZE_THRESHOLD_TOKENS: usize = 20_000;
|
||||
|
||||
/// Characters of the raw payload to surface in the placeholder preview.
|
||||
/// Enough for the sub-agent to recognise the shape (JSON keys, first
|
||||
/// record) and often small enough to answer trivial questions without a
|
||||
/// follow-up `extract_from_result` call.
|
||||
const HANDOFF_PREVIEW_CHARS: usize = 1500;
|
||||
|
||||
/// Maximum entries per session. Bounded to keep memory use predictable
|
||||
/// on long-running sub-agents that might call many large tools. When
|
||||
/// over capacity we evict the oldest entry (FIFO); callers see "no
|
||||
/// cached result" for evicted ids and can either re-run the tool or
|
||||
/// ask the user/orchestrator to narrow the request.
|
||||
const HANDOFF_MAX_ENTRIES: usize = 8;
|
||||
|
||||
/// Per-spawn cache of oversized tool payloads. One instance is built
|
||||
/// at the top of [`run_typed_mode`] and shared (via `Arc`) with both
|
||||
/// the inner tool-call loop (writes) and the `extract_from_result`
|
||||
/// tool (reads).
|
||||
#[derive(Default)]
|
||||
struct ResultHandoffCache {
|
||||
inner: StdMutex<HandoffInner>,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct HandoffInner {
|
||||
/// FIFO of inserted ids, used for eviction.
|
||||
order: Vec<String>,
|
||||
/// Content by id.
|
||||
entries: HashMap<String, CachedResult>,
|
||||
/// Monotonic counter for id generation within the session.
|
||||
next_id: u64,
|
||||
}
|
||||
|
||||
struct CachedResult {
|
||||
tool_name: String,
|
||||
content: String,
|
||||
}
|
||||
|
||||
impl ResultHandoffCache {
|
||||
fn new() -> Self {
|
||||
Self::default()
|
||||
}
|
||||
|
||||
/// Stash a payload and return a stable, short, grep-friendly id.
|
||||
fn store(&self, tool_name: String, content: String) -> String {
|
||||
let mut g = match self.inner.lock() {
|
||||
Ok(g) => g,
|
||||
Err(poisoned) => poisoned.into_inner(),
|
||||
};
|
||||
g.next_id = g.next_id.saturating_add(1);
|
||||
let id = format!("res_{:x}", g.next_id);
|
||||
g.order.push(id.clone());
|
||||
g.entries
|
||||
.insert(id.clone(), CachedResult { tool_name, content });
|
||||
while g.order.len() > HANDOFF_MAX_ENTRIES {
|
||||
let evicted = g.order.remove(0);
|
||||
g.entries.remove(&evicted);
|
||||
}
|
||||
id
|
||||
}
|
||||
|
||||
fn get(&self, result_id: &str) -> Option<CachedResult> {
|
||||
let g = self.inner.lock().ok()?;
|
||||
g.entries.get(result_id).map(|r| CachedResult {
|
||||
tool_name: r.tool_name.clone(),
|
||||
content: r.content.clone(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// Build the placeholder text that replaces an oversized tool result in
|
||||
/// the sub-agent's history. Shows the payload size (estimated tokens
|
||||
/// and raw bytes), a preview, and a call shape for the
|
||||
/// `extract_from_result` tool. The sub-agent decides whether to answer
|
||||
/// from the preview or dispatch the extractor.
|
||||
///
|
||||
/// Token count is estimated at ~4 chars/token (same heuristic as the
|
||||
/// trigger threshold in `HANDOFF_OVERSIZE_THRESHOLD_TOKENS`), so the
|
||||
/// unit the sub-agent sees matches the unit the runtime used to decide
|
||||
/// to hand off in the first place.
|
||||
fn build_handoff_placeholder(tool_name: &str, result_id: &str, raw: &str) -> String {
|
||||
let preview: String = raw.chars().take(HANDOFF_PREVIEW_CHARS).collect();
|
||||
let raw_tokens = raw.len().div_ceil(4);
|
||||
format!(
|
||||
"[oversized tool output: {raw_tokens} tokens ({raw_bytes} bytes) — stashed as result_id=\"{result_id}\"]\n\
|
||||
Preview (first {preview_chars} chars):\n{preview}\n\n\
|
||||
If the preview does not answer your task, call:\n\
|
||||
extract_from_result(result_id=\"{result_id}\", query=\"<specific question>\")\n\
|
||||
Good queries name the exact fields/identifiers you need \
|
||||
(e.g. \"subject and sender of the 5 most recent messages\"). \
|
||||
Tool: {tool_name}",
|
||||
raw_bytes = raw.len(),
|
||||
preview_chars = preview.chars().count(),
|
||||
)
|
||||
}
|
||||
|
||||
/// Sub-agent-side tool that answers a targeted query against a payload
|
||||
/// previously stashed via [`build_handoff_placeholder`]. Internally
|
||||
/// dispatches the `summarizer` sub-agent with the cached payload + the
|
||||
/// caller's query as the extraction contract.
|
||||
struct ExtractFromResultTool {
|
||||
cache: Arc<ResultHandoffCache>,
|
||||
summarizer_def: AgentDefinition,
|
||||
}
|
||||
|
||||
impl ExtractFromResultTool {
|
||||
fn new(cache: Arc<ResultHandoffCache>, summarizer_def: AgentDefinition) -> Self {
|
||||
Self {
|
||||
cache,
|
||||
summarizer_def,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Tool for ExtractFromResultTool {
|
||||
fn name(&self) -> &str {
|
||||
"extract_from_result"
|
||||
}
|
||||
|
||||
fn description(&self) -> &str {
|
||||
"Answer a targeted question against an oversized tool output that was \
|
||||
stashed under a `result_id` handle. Use this when a previous tool call \
|
||||
returned a placeholder like `result_id=\"res_1\"` because its raw output \
|
||||
was too large to show inline. Pass the handle plus a natural-language \
|
||||
`query` naming the exact facts/identifiers you need; returns only the \
|
||||
extracted answer, not the full payload. Multiple queries against the \
|
||||
same `result_id` are allowed — each one is independent."
|
||||
}
|
||||
|
||||
fn parameters_schema(&self) -> serde_json::Value {
|
||||
serde_json::json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"result_id": {
|
||||
"type": "string",
|
||||
"description": "The handle emitted in the oversized tool output placeholder (e.g. `res_1`)."
|
||||
},
|
||||
"query": {
|
||||
"type": "string",
|
||||
"description": "Natural-language question naming the exact facts or identifiers to extract. Be specific."
|
||||
}
|
||||
},
|
||||
"required": ["result_id", "query"]
|
||||
})
|
||||
}
|
||||
|
||||
fn category(&self) -> ToolCategory {
|
||||
ToolCategory::System
|
||||
}
|
||||
|
||||
async fn execute(&self, args: serde_json::Value) -> anyhow::Result<ToolResult> {
|
||||
let result_id = args.get("result_id").and_then(|v| v.as_str()).unwrap_or("");
|
||||
let query = args.get("query").and_then(|v| v.as_str()).unwrap_or("");
|
||||
|
||||
if result_id.is_empty() || query.is_empty() {
|
||||
return Ok(ToolResult::error(
|
||||
"extract_from_result requires non-empty `result_id` and `query`.",
|
||||
));
|
||||
}
|
||||
|
||||
let cached = match self.cache.get(result_id) {
|
||||
Some(c) => c,
|
||||
None => {
|
||||
return Ok(ToolResult::error(format!(
|
||||
"No cached result found for id '{result_id}'. The handle may have been evicted (cache holds the {HANDOFF_MAX_ENTRIES} most recent entries). Re-run the original tool to get a fresh handle."
|
||||
)));
|
||||
}
|
||||
};
|
||||
|
||||
// Fast path: payload fits in a single summarizer turn.
|
||||
if cached.content.len() <= SUMMARIZER_CHUNK_CHAR_BUDGET {
|
||||
tracing::debug!(
|
||||
tool = %cached.tool_name,
|
||||
bytes = cached.content.len(),
|
||||
"[extract_from_result] single-shot extraction"
|
||||
);
|
||||
return self
|
||||
.extract_single_shot(&cached.tool_name, &cached.content, query)
|
||||
.await;
|
||||
}
|
||||
|
||||
// Slow path: chunk + parallel map. A single summarizer call on a
|
||||
// payload large enough to need the handoff (hundreds of KB
|
||||
// common for Gmail/Notion list operations) risks either (a)
|
||||
// overflowing the summarizer's own context window, or (b)
|
||||
// getting a low-quality single-pass summary that misses facts
|
||||
// near the tail. Splitting into budgeted chunks and running
|
||||
// them in parallel keeps each summarizer turn under its
|
||||
// context budget and usually finishes faster than a sequential
|
||||
// single-shot call on the whole blob.
|
||||
//
|
||||
// No reduce stage: per-chunk summaries are concatenated in
|
||||
// original chunk order. The reduce LLM call previously used
|
||||
// here added latency (often the slowest single turn of the
|
||||
// whole pipeline) and was the single point of failure when
|
||||
// the upstream provider hung — a hung reduce could burn
|
||||
// minutes before giving up. For listing/extraction queries
|
||||
// concatenation is equivalent; for top-N / global-ordering
|
||||
// queries the caller can post-process.
|
||||
let chunks = chunk_content(&cached.content, SUMMARIZER_CHUNK_CHAR_BUDGET);
|
||||
tracing::info!(
|
||||
tool = %cached.tool_name,
|
||||
total_bytes = cached.content.len(),
|
||||
chunk_count = chunks.len(),
|
||||
chunk_budget = SUMMARIZER_CHUNK_CHAR_BUDGET,
|
||||
"[extract_from_result] chunked extraction"
|
||||
);
|
||||
|
||||
// Map stage: each chunk extracts items matching `query` from
|
||||
// ITS OWN slice only. Dispatched with bounded concurrency —
|
||||
// `buffer_unordered(MAP_CONCURRENCY)` keeps at most N summarizer
|
||||
// calls in flight at any time. Fully parallel `join_all` was
|
||||
// generating 504-gateway-timeout storms from the staging
|
||||
// proxy when 7+ concurrent summarizer calls piled onto the
|
||||
// upstream; batching at 3 trades some wall-clock time for
|
||||
// reliability. `run_subagent` is isolated per call (fresh
|
||||
// history, independent summarizer instance). Callers share
|
||||
// the same parent context.
|
||||
const MAP_CONCURRENCY: usize = 3;
|
||||
let total_chunks = chunks.len();
|
||||
|
||||
// Consume `chunks` with `into_iter` so each async block owns
|
||||
// its `String` — `buffer_unordered` polls the stream lazily
|
||||
// and needs futures with no borrows into the enclosing scope.
|
||||
let map_futures = chunks.into_iter().enumerate().map(|(i, chunk)| {
|
||||
let summarizer_def = self.summarizer_def.clone();
|
||||
let tool_name = cached.tool_name.clone();
|
||||
let query = query.to_string();
|
||||
async move {
|
||||
let prompt = format!(
|
||||
"Tool name: {tool_name}\nChunk {idx} of {total}\n\n\
|
||||
Query: {query}\n\n\
|
||||
This is one slice of a larger tool output. Extract ONLY \
|
||||
items in THIS slice that match the query. Preserve \
|
||||
identifiers verbatim (ids, urls, emails, timestamps). \
|
||||
Return an empty string if nothing in this slice is \
|
||||
relevant. Be concise — no preamble, no commentary on \
|
||||
other slices.\n\n\
|
||||
--- BEGIN SLICE ---\n{chunk}\n--- END SLICE ---",
|
||||
idx = i + 1,
|
||||
total = total_chunks,
|
||||
);
|
||||
(
|
||||
i,
|
||||
run_subagent(&summarizer_def, &prompt, SubagentRunOptions::default()).await,
|
||||
)
|
||||
}
|
||||
});
|
||||
|
||||
use futures::stream::StreamExt;
|
||||
let mut map_results: Vec<(usize, _)> = futures::stream::iter(map_futures)
|
||||
.buffer_unordered(MAP_CONCURRENCY)
|
||||
.collect()
|
||||
.await;
|
||||
// `buffer_unordered` yields futures in completion order; restore
|
||||
// original chunk order so the concatenated output matches the
|
||||
// natural ordering of the underlying tool result (e.g. Notion's
|
||||
// reverse-chrono page list).
|
||||
map_results.sort_by_key(|(i, _)| *i);
|
||||
|
||||
let partials: Vec<String> = map_results
|
||||
.into_iter()
|
||||
.filter_map(|(i, r)| match r {
|
||||
Ok(outcome) => {
|
||||
let trimmed = outcome.output.trim();
|
||||
if trimmed.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(trimmed.to_string())
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
chunk_idx = i,
|
||||
error = %e,
|
||||
"[extract_from_result] map-stage summarizer call failed; dropping partial"
|
||||
);
|
||||
None
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
|
||||
if partials.is_empty() {
|
||||
return Ok(ToolResult::error(
|
||||
"extract_from_result: no matching content found across any chunk",
|
||||
));
|
||||
}
|
||||
|
||||
// Concatenate per-chunk summaries in original chunk order.
|
||||
// `join` with a single partial yields it unchanged (no trailing
|
||||
// separator), so no special-case is needed.
|
||||
Ok(ToolResult::success(partials.join("\n\n---\n\n")))
|
||||
}
|
||||
}
|
||||
|
||||
impl ExtractFromResultTool {
|
||||
async fn extract_single_shot(
|
||||
&self,
|
||||
tool_name: &str,
|
||||
content: &str,
|
||||
query: &str,
|
||||
) -> anyhow::Result<ToolResult> {
|
||||
let prompt = format!(
|
||||
"Tool name: {tool_name}\n\nQuery: {query}\n\n\
|
||||
Raw tool output — extract ONLY the information the query asks for. \
|
||||
Preserve identifiers verbatim (ids, urls, emails, timestamps). \
|
||||
Return a compact, direct answer, no preamble.\n\n\
|
||||
--- BEGIN ---\n{content}\n--- END ---",
|
||||
);
|
||||
|
||||
match run_subagent(&self.summarizer_def, &prompt, SubagentRunOptions::default()).await {
|
||||
Ok(run) => {
|
||||
let trimmed = run.output.trim();
|
||||
if trimmed.is_empty() {
|
||||
Ok(ToolResult::error(
|
||||
"extract_from_result: summarizer returned an empty response",
|
||||
))
|
||||
} else {
|
||||
Ok(ToolResult::success(trimmed.to_string()))
|
||||
}
|
||||
}
|
||||
Err(e) => Ok(ToolResult::error(format!(
|
||||
"extract_from_result: summarizer dispatch failed: {e}"
|
||||
))),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Char budget per summarizer call. Chosen so a single chunk + prompt
|
||||
/// scaffolding + output stays well below the summarization model's
|
||||
/// context window (~196k tokens) — at ~4 chars/token that leaves
|
||||
/// comfortable headroom for the extraction contract and response.
|
||||
const SUMMARIZER_CHUNK_CHAR_BUDGET: usize = 60_000;
|
||||
|
||||
/// Split `content` into chunks no larger than `budget` bytes, breaking
|
||||
/// at natural boundaries (blank lines, then single newlines) so the
|
||||
/// summarizer rarely sees a structure torn mid-record. Falls back to
|
||||
/// char-safe slicing for pathological single-line inputs.
|
||||
/// Strip common noise from tool outputs before they're stashed or chunked.
|
||||
///
|
||||
/// Agent tools frequently return raw HTML email bodies, inline SVG, base64
|
||||
/// data URIs, CSS/JS blocks, and collapsed whitespace — all of which bloat
|
||||
/// the handoff cache and waste summarizer context on tokens that carry
|
||||
/// zero semantic value for most extraction queries. Cleaning before the
|
||||
/// oversize check means (a) some payloads drop below threshold entirely
|
||||
/// and skip the extract pipeline, (b) chunked payloads fit more real
|
||||
/// content per chunk, and (c) summarizers see clean text instead of
|
||||
/// parsing around markup.
|
||||
///
|
||||
/// Applied generically to every tool output — no per-tool gating. The
|
||||
/// patterns below target universally-noisy markup; non-HTML outputs
|
||||
/// (plain JSON, plain text) are left essentially untouched because none
|
||||
/// of these regexes match.
|
||||
fn clean_tool_output(content: &str) -> String {
|
||||
// Block-level containers with entirely useless payloads — match lazily
|
||||
// across lines, case-insensitive. `(?s)` enables `.` matching `\n`.
|
||||
static SCRIPT_RE: LazyLock<Regex> =
|
||||
LazyLock::new(|| Regex::new(r"(?is)<script\b[^>]*>.*?</script\s*>").unwrap());
|
||||
static STYLE_RE: LazyLock<Regex> =
|
||||
LazyLock::new(|| Regex::new(r"(?is)<style\b[^>]*>.*?</style\s*>").unwrap());
|
||||
static SVG_RE: LazyLock<Regex> =
|
||||
LazyLock::new(|| Regex::new(r"(?is)<svg\b[^>]*>.*?</svg\s*>").unwrap());
|
||||
static HTML_COMMENT_RE: LazyLock<Regex> =
|
||||
LazyLock::new(|| Regex::new(r"(?s)<!--.*?-->").unwrap());
|
||||
// `data:<mime>;base64,<...>` inline payloads — keep the agent aware
|
||||
// an image/asset was there, but drop the bytes.
|
||||
static DATA_URI_RE: LazyLock<Regex> =
|
||||
LazyLock::new(|| Regex::new(r"(?i)data:[a-z0-9.+\-/]+;base64,[A-Za-z0-9+/=]+").unwrap());
|
||||
// Everything remaining that still looks like an HTML tag — strip the
|
||||
// tag but keep the text content. Deliberately greedy across attributes
|
||||
// but not across `>` so we don't eat inter-tag content.
|
||||
static HTML_TAG_RE: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"<[^>]+>").unwrap());
|
||||
// Runs of whitespace → single space; collapse vertical bloat.
|
||||
static WS_RE: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"[ \t\f\v]+").unwrap());
|
||||
static BLANK_LINE_RE: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"\n{3,}").unwrap());
|
||||
|
||||
let cleaned = SCRIPT_RE.replace_all(content, "");
|
||||
let cleaned = STYLE_RE.replace_all(&cleaned, "");
|
||||
let cleaned = SVG_RE.replace_all(&cleaned, "[svg]");
|
||||
let cleaned = HTML_COMMENT_RE.replace_all(&cleaned, "");
|
||||
let cleaned = DATA_URI_RE.replace_all(&cleaned, "[data-uri]");
|
||||
let cleaned = HTML_TAG_RE.replace_all(&cleaned, "");
|
||||
let cleaned = WS_RE.replace_all(&cleaned, " ");
|
||||
let cleaned = BLANK_LINE_RE.replace_all(&cleaned, "\n\n");
|
||||
cleaned.trim().to_string()
|
||||
}
|
||||
|
||||
fn chunk_content(content: &str, budget: usize) -> Vec<String> {
|
||||
if content.len() <= budget {
|
||||
return vec![content.to_string()];
|
||||
}
|
||||
|
||||
let mut chunks: Vec<String> = Vec::new();
|
||||
let mut current = String::with_capacity(budget.min(content.len()));
|
||||
|
||||
let flush = |current: &mut String, chunks: &mut Vec<String>| {
|
||||
if !current.is_empty() {
|
||||
chunks.push(std::mem::take(current));
|
||||
}
|
||||
};
|
||||
|
||||
for line in content.lines() {
|
||||
let projected = current.len() + line.len() + 1;
|
||||
if projected > budget && !current.is_empty() {
|
||||
flush(&mut current, &mut chunks);
|
||||
}
|
||||
if line.len() > budget {
|
||||
// Single line exceeds budget (e.g. JSON with no formatting).
|
||||
// Emit any pending content, then slice the line at char
|
||||
// boundaries so we don't panic on multi-byte chars.
|
||||
flush(&mut current, &mut chunks);
|
||||
let mut remaining = line;
|
||||
while !remaining.is_empty() {
|
||||
let mut cut = budget.min(remaining.len());
|
||||
while cut > 0 && !remaining.is_char_boundary(cut) {
|
||||
cut -= 1;
|
||||
}
|
||||
if cut == 0 {
|
||||
// Degenerate case — shouldn't happen for normal
|
||||
// text. Take the entire remaining line to avoid
|
||||
// an infinite loop.
|
||||
chunks.push(remaining.to_string());
|
||||
break;
|
||||
}
|
||||
chunks.push(remaining[..cut].to_string());
|
||||
remaining = &remaining[cut..];
|
||||
}
|
||||
} else {
|
||||
current.push_str(line);
|
||||
current.push('\n');
|
||||
}
|
||||
}
|
||||
flush(&mut current, &mut chunks);
|
||||
|
||||
chunks
|
||||
}
|
||||
|
||||
/// Tight top-K ceiling for toolkits whose per-action JSON schemas are
|
||||
/// dense enough to blow through either Fireworks' 65 535-rule grammar
|
||||
/// cap (native mode) or the 196 607-token context cap (text mode) even
|
||||
/// before any tool results land in history. Determined empirically from
|
||||
/// the fixture dumps under `tests/fixtures/composio_*.json` and real
|
||||
/// staging failures — see the trace where Gmail at top-K=25 produced
|
||||
/// a 276k-token iter-1 prompt.
|
||||
const HEAVY_SCHEMA_TOOLKITS: &[&str] = &[
|
||||
"gmail",
|
||||
"notion",
|
||||
"github",
|
||||
"salesforce",
|
||||
"hubspot",
|
||||
"googledrive",
|
||||
"googlesheets",
|
||||
"googledocs",
|
||||
"microsoftteams",
|
||||
];
|
||||
|
||||
const TOOL_FILTER_TOP_K_DEFAULT: usize = 25;
|
||||
const TOOL_FILTER_TOP_K_HEAVY: usize = 12;
|
||||
|
||||
/// Pick a top-K budget for the fuzzy filter based on how dense the
|
||||
/// toolkit's action schemas tend to be. Match is case-insensitive so
|
||||
/// we don't care whether the caller passed `"Gmail"` or `"gmail"`.
|
||||
fn top_k_for_toolkit(toolkit: &str) -> usize {
|
||||
if HEAVY_SCHEMA_TOOLKITS
|
||||
.iter()
|
||||
.any(|t| t.eq_ignore_ascii_case(toolkit))
|
||||
{
|
||||
TOOL_FILTER_TOP_K_HEAVY
|
||||
} else {
|
||||
TOOL_FILTER_TOP_K_DEFAULT
|
||||
}
|
||||
}
|
||||
|
||||
/// Format a set of `ToolSpec`s as an XML tool-use protocol block
|
||||
/// appended to the system prompt in text mode. Mirrors
|
||||
/// [`crate::openhuman::agent::dispatcher::XmlToolDispatcher::prompt_instructions`]
|
||||
/// — same `<tool_call>{…}</tool_call>` format so the existing
|
||||
/// `parse_tool_calls` helper understands what the model emits.
|
||||
///
|
||||
/// Per-parameter rendering is intentionally **compact**: name, type, a
|
||||
/// "required" marker, and a short one-line description if present. We
|
||||
/// do **not** serialise the full JSON schema. Composio/Fireworks action
|
||||
/// schemas for toolkits like Gmail or Notion run multiple KB each —
|
||||
/// embedding them verbatim blows up the prompt past the model's
|
||||
/// context window (282k+ tokens for 26 Gmail tools vs a 196k cap).
|
||||
/// The compact listing keeps the model informed enough to call tools
|
||||
/// correctly while staying within budget. If the model needs deeper
|
||||
/// schema detail it can surface the error and the orchestrator will
|
||||
/// clarify on the next turn.
|
||||
fn build_text_mode_tool_instructions(specs: &[ToolSpec]) -> String {
|
||||
use std::fmt::Write as _;
|
||||
|
||||
let mut out = String::new();
|
||||
out.push_str("## Tool Use Protocol\n\n");
|
||||
out.push_str(
|
||||
"To use a tool, wrap a JSON object in <tool_call></tool_call> tags. \
|
||||
Do not nest tags. Emit one tag per call; you can emit multiple tags \
|
||||
in the same response if you need to run calls in parallel.\n\n",
|
||||
);
|
||||
out.push_str(
|
||||
"```\n<tool_call>\n{\"name\": \"tool_name\", \"arguments\": {\"param\": \"value\"}}\n</tool_call>\n```\n\n",
|
||||
);
|
||||
out.push_str("### Available Tools\n\n");
|
||||
for spec in specs {
|
||||
let _ = writeln!(
|
||||
out,
|
||||
"- **{}**: {}",
|
||||
spec.name,
|
||||
first_line_truncated(&spec.description, 120)
|
||||
);
|
||||
let params_line = summarise_parameters(&spec.parameters);
|
||||
if !params_line.is_empty() {
|
||||
let _ = writeln!(out, " Parameters: {}", params_line);
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// Render a JSON-schema `parameters` object as a single-line,
|
||||
/// ultra-compact parameter list — `*name: type, optName: type` for each
|
||||
/// top-level property (leading `*` marks required). Deeply nested
|
||||
/// shapes, enums, descriptions, and examples are all dropped.
|
||||
///
|
||||
/// Kept intentionally terse: Composio action schemas routinely contain
|
||||
/// per-parameter descriptions several hundred tokens long, so even a
|
||||
/// "short description" per param balloons to tens of thousands of
|
||||
/// tokens across a 27-tool skills_agent toolkit and pushes the prompt
|
||||
/// past the 196 607-token context window. The model can infer usage
|
||||
/// from the parameter names + the tool's overall description; any
|
||||
/// validation mismatch surfaces at call time and the orchestrator can
|
||||
/// course-correct on the next turn.
|
||||
fn summarise_parameters(params: &serde_json::Value) -> String {
|
||||
let Some(props) = params.get("properties").and_then(|v| v.as_object()) else {
|
||||
return String::new();
|
||||
};
|
||||
let required: std::collections::HashSet<&str> = params
|
||||
.get("required")
|
||||
.and_then(|v| v.as_array())
|
||||
.map(|arr| arr.iter().filter_map(|v| v.as_str()).collect())
|
||||
.unwrap_or_default();
|
||||
|
||||
let mut parts: Vec<String> = Vec::with_capacity(props.len());
|
||||
for (name, schema) in props {
|
||||
let ty = schema.get("type").and_then(|v| v.as_str()).unwrap_or("any");
|
||||
let marker = if required.contains(name.as_str()) {
|
||||
"*"
|
||||
} else {
|
||||
""
|
||||
};
|
||||
parts.push(format!("{marker}{name}:{ty}"));
|
||||
}
|
||||
parts.join(", ")
|
||||
}
|
||||
|
||||
/// Return the first line of `s`, trimmed and truncated to `max_chars`
|
||||
/// with a trailing ellipsis when it overflows. Used to keep
|
||||
/// tool/parameter descriptions on a single grep-friendly line.
|
||||
fn first_line_truncated(s: &str, max_chars: usize) -> String {
|
||||
let first = s.lines().next().unwrap_or("").trim();
|
||||
if first.chars().count() <= max_chars {
|
||||
first.to_string()
|
||||
} else {
|
||||
let truncated: String = first.chars().take(max_chars).collect();
|
||||
format!("{truncated}…")
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Tool filtering
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
@@ -927,6 +1779,7 @@ mod tests {
|
||||
disallowed_tools: vec![],
|
||||
skill_filter: None,
|
||||
category_filter: None,
|
||||
extra_tools: vec![],
|
||||
max_iterations: 5,
|
||||
timeout_secs: None,
|
||||
sandbox_mode: super::super::definition::SandboxMode::None,
|
||||
|
||||
@@ -127,6 +127,7 @@ async fn run_tool_call_loop_returns_structured_error_for_non_vision_provider() {
|
||||
None,
|
||||
&[],
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect_err("provider without vision support should fail");
|
||||
@@ -171,6 +172,7 @@ async fn run_tool_call_loop_rejects_oversized_image_payload() {
|
||||
None,
|
||||
&[],
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect_err("oversized payload must fail");
|
||||
@@ -209,6 +211,7 @@ async fn run_tool_call_loop_accepts_valid_multimodal_request_flow() {
|
||||
None,
|
||||
&[],
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("valid multimodal payload should pass");
|
||||
|
||||
@@ -13,6 +13,7 @@ use std::io::Write as _;
|
||||
|
||||
use super::credentials::scrub_credentials;
|
||||
use super::parse::{build_native_assistant_history, parse_structured_tool_calls, parse_tool_calls};
|
||||
use super::payload_summarizer::PayloadSummarizer;
|
||||
use crate::openhuman::context::guard::{ContextCheckResult, ContextGuard};
|
||||
|
||||
/// Minimum characters per chunk when relaying LLM text to a streaming draft.
|
||||
@@ -41,6 +42,7 @@ pub(crate) async fn agent_turn(
|
||||
silent: bool,
|
||||
multimodal_config: &crate::openhuman::config::MultimodalConfig,
|
||||
max_tool_iterations: usize,
|
||||
payload_summarizer: Option<&dyn PayloadSummarizer>,
|
||||
) -> Result<String> {
|
||||
run_tool_call_loop(
|
||||
provider,
|
||||
@@ -58,6 +60,7 @@ pub(crate) async fn agent_turn(
|
||||
None,
|
||||
&[],
|
||||
None,
|
||||
payload_summarizer,
|
||||
)
|
||||
.await
|
||||
}
|
||||
@@ -107,6 +110,7 @@ pub(crate) async fn run_tool_call_loop(
|
||||
visible_tool_names: Option<&HashSet<String>>,
|
||||
extra_tools: &[Box<dyn Tool>],
|
||||
on_progress: Option<tokio::sync::mpsc::Sender<AgentProgress>>,
|
||||
payload_summarizer: Option<&dyn PayloadSummarizer>,
|
||||
) -> Result<String> {
|
||||
let max_iterations = if max_tool_iterations == 0 {
|
||||
DEFAULT_MAX_TOOL_ITERATIONS
|
||||
@@ -552,7 +556,43 @@ pub(crate) async fn run_tool_call_loop(
|
||||
output_len = output.len(),
|
||||
"[agent_loop] tool succeeded"
|
||||
);
|
||||
(scrub_credentials(&output), true)
|
||||
let mut scrubbed = scrub_credentials(&output);
|
||||
if let Some(summarizer) = payload_summarizer {
|
||||
log::debug!(
|
||||
"[agent_loop] payload_summarizer intercepting tool={} bytes={}",
|
||||
call.name,
|
||||
scrubbed.len()
|
||||
);
|
||||
match summarizer
|
||||
.maybe_summarize(&call.name, None, &scrubbed)
|
||||
.await
|
||||
{
|
||||
Ok(Some(payload)) => {
|
||||
log::info!(
|
||||
"[agent_loop] payload_summarizer compressed tool={} {}->{} bytes",
|
||||
call.name,
|
||||
payload.original_bytes,
|
||||
payload.summary_bytes
|
||||
);
|
||||
scrubbed = payload.summary;
|
||||
}
|
||||
Ok(None) => {
|
||||
log::debug!(
|
||||
"[agent_loop] payload_summarizer pass-through tool={} bytes={}",
|
||||
call.name,
|
||||
scrubbed.len()
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
log::warn!(
|
||||
"[agent_loop] payload_summarizer error tool={} err={} (passing raw payload through)",
|
||||
call.name,
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
(scrubbed, true)
|
||||
} else {
|
||||
tracing::warn!(
|
||||
iteration,
|
||||
@@ -779,6 +819,124 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
/// Tool that emits a large payload (~150 KB), used to exercise the
|
||||
/// payload-summarizer interception path in the integration test
|
||||
/// below.
|
||||
struct BigPayloadTool;
|
||||
|
||||
#[async_trait]
|
||||
impl Tool for BigPayloadTool {
|
||||
fn name(&self) -> &str {
|
||||
"big_payload"
|
||||
}
|
||||
|
||||
fn description(&self) -> &str {
|
||||
"emits a 150 KB payload to trigger summarization"
|
||||
}
|
||||
|
||||
fn parameters_schema(&self) -> serde_json::Value {
|
||||
serde_json::json!({"type":"object"})
|
||||
}
|
||||
|
||||
async fn execute(&self, _args: serde_json::Value) -> Result<ToolResult> {
|
||||
// 150 KB of payload — well above the 100 KB default threshold.
|
||||
Ok(ToolResult::success("X".repeat(150_000)))
|
||||
}
|
||||
}
|
||||
|
||||
/// Mock summarizer that always returns a fixed compressed string,
|
||||
/// used to verify that [`run_tool_call_loop`] swaps the raw tool
|
||||
/// output for the summary before pushing it into history.
|
||||
struct MockSummarizer {
|
||||
summary: String,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl super::super::payload_summarizer::PayloadSummarizer for MockSummarizer {
|
||||
async fn maybe_summarize(
|
||||
&self,
|
||||
_tool_name: &str,
|
||||
_parent_task_hint: Option<&str>,
|
||||
raw: &str,
|
||||
) -> Result<Option<super::super::payload_summarizer::SummarizedPayload>> {
|
||||
Ok(Some(super::super::payload_summarizer::SummarizedPayload {
|
||||
summary: self.summary.clone(),
|
||||
original_bytes: raw.len(),
|
||||
summary_bytes: self.summary.len(),
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn run_tool_call_loop_intercepts_oversized_tool_results_via_summarizer() {
|
||||
// Provider scripts a single tool call to `big_payload`, then a
|
||||
// final "done" message after the tool result lands in history.
|
||||
let provider = ScriptedProvider {
|
||||
responses: Mutex::new(vec![
|
||||
Ok(ChatResponse {
|
||||
text: Some(
|
||||
"<tool_call>{\"name\":\"big_payload\",\"arguments\":{}}</tool_call>".into(),
|
||||
),
|
||||
tool_calls: vec![],
|
||||
usage: None,
|
||||
}),
|
||||
Ok(ChatResponse {
|
||||
text: Some("done".into()),
|
||||
tool_calls: vec![],
|
||||
usage: None,
|
||||
}),
|
||||
]),
|
||||
native_tools: false,
|
||||
vision: false,
|
||||
};
|
||||
let mut history = vec![ChatMessage::user("dump the data")];
|
||||
let tools: Vec<Box<dyn Tool>> = vec![Box::new(BigPayloadTool)];
|
||||
let summarizer = MockSummarizer {
|
||||
summary: "compressed-summary-marker".to_string(),
|
||||
};
|
||||
|
||||
let result = run_tool_call_loop(
|
||||
&provider,
|
||||
&mut history,
|
||||
&tools,
|
||||
"test-provider",
|
||||
"model",
|
||||
0.0,
|
||||
true,
|
||||
None,
|
||||
"channel",
|
||||
&crate::openhuman::config::MultimodalConfig::default(),
|
||||
2,
|
||||
None,
|
||||
None,
|
||||
&[],
|
||||
None,
|
||||
Some(&summarizer),
|
||||
)
|
||||
.await
|
||||
.expect("loop with summarizer should succeed");
|
||||
|
||||
assert_eq!(result, "done");
|
||||
|
||||
// The summarized marker should be present in the appended
|
||||
// tool-results message; the raw 150 KB blob of 'X' should NOT.
|
||||
let tool_results = history
|
||||
.iter()
|
||||
.find(|msg| msg.role == "user" && msg.content.contains("[Tool results]"))
|
||||
.expect("tool results should be appended");
|
||||
assert!(
|
||||
tool_results.content.contains("compressed-summary-marker"),
|
||||
"summarizer output should replace the raw payload in history"
|
||||
);
|
||||
// 150 KB of "X" is much larger than the summary; if it slipped
|
||||
// through, the message body would be enormous.
|
||||
assert!(
|
||||
tool_results.content.len() < 10_000,
|
||||
"raw 150 KB payload must not appear in history (got {} bytes)",
|
||||
tool_results.content.len()
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn run_tool_call_loop_rejects_vision_markers_for_non_vision_provider() {
|
||||
let provider = ScriptedProvider {
|
||||
@@ -804,6 +962,7 @@ mod tests {
|
||||
None,
|
||||
&[],
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect_err("vision markers should be rejected");
|
||||
@@ -841,6 +1000,7 @@ mod tests {
|
||||
None,
|
||||
&[],
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("final text should succeed");
|
||||
@@ -893,6 +1053,7 @@ mod tests {
|
||||
None,
|
||||
&[],
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("loop should recover after denial");
|
||||
@@ -948,6 +1109,7 @@ mod tests {
|
||||
None,
|
||||
&[],
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("native tool flow should succeed");
|
||||
@@ -1006,6 +1168,7 @@ mod tests {
|
||||
None,
|
||||
&[],
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("non-cli channels should auto-approve supervised tools");
|
||||
@@ -1057,6 +1220,7 @@ mod tests {
|
||||
None,
|
||||
&[],
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("default iteration fallback should still succeed");
|
||||
@@ -1112,6 +1276,7 @@ mod tests {
|
||||
None,
|
||||
&[],
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("loop should recover after tool errors");
|
||||
@@ -1151,6 +1316,7 @@ mod tests {
|
||||
None,
|
||||
&[],
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect_err("provider error path should fail");
|
||||
@@ -1183,6 +1349,7 @@ mod tests {
|
||||
None,
|
||||
&[],
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect_err("loop should stop after configured iterations");
|
||||
|
||||
@@ -532,6 +532,7 @@ mod scoping_tests {
|
||||
disallowed_tools: vec![],
|
||||
skill_filter: None,
|
||||
category_filter: None,
|
||||
extra_tools: vec![],
|
||||
max_iterations: 8,
|
||||
timeout_secs: None,
|
||||
sandbox_mode: SandboxMode::None,
|
||||
|
||||
@@ -72,6 +72,37 @@ pub struct ContextConfig {
|
||||
#[serde(default = "default_tool_result_budget_bytes")]
|
||||
pub tool_result_budget_bytes: usize,
|
||||
|
||||
/// Tool results larger than this **token** count trigger the
|
||||
/// `summarizer` sub-agent (orchestrator session only). The summarizer
|
||||
/// compresses the payload into a dense note that preserves
|
||||
/// identifiers and key facts, and the compressed summary replaces
|
||||
/// the raw payload before it enters agent history. Set to `0` to
|
||||
/// disable summarization entirely. Default: `500_000` tokens.
|
||||
///
|
||||
/// Token count is estimated as `chars / 4` (the same heuristic used
|
||||
/// by `tree_summarizer::estimate_tokens`). Pairs with
|
||||
/// [`Self::summarizer_max_payload_tokens`] which caps the upper end
|
||||
/// (paying for an LLM call on a multi-million-token blob makes no
|
||||
/// economic sense, so above the cap the existing
|
||||
/// [`Self::tool_result_budget_bytes`] truncation handles it instead).
|
||||
#[serde(
|
||||
default = "default_summarizer_payload_threshold_tokens",
|
||||
alias = "summarizer_payload_threshold_bytes"
|
||||
)]
|
||||
pub summarizer_payload_threshold_tokens: usize,
|
||||
|
||||
/// Hard cap on payload size (in **tokens**) above which summarization
|
||||
/// is skipped entirely and the existing
|
||||
/// [`Self::tool_result_budget_bytes`] truncation path takes over.
|
||||
/// Default: `2_000_000` tokens (above the context window of every
|
||||
/// model we ship against — a payload this big can't be summarized
|
||||
/// cost-effectively).
|
||||
#[serde(
|
||||
default = "default_summarizer_max_payload_tokens",
|
||||
alias = "summarizer_max_payload_bytes"
|
||||
)]
|
||||
pub summarizer_max_payload_tokens: usize,
|
||||
|
||||
/// Session-memory extraction thresholds (stage 5 of the pipeline).
|
||||
#[serde(default)]
|
||||
pub session_memory: SessionMemoryConfig,
|
||||
@@ -112,6 +143,14 @@ fn default_tool_result_budget_bytes() -> usize {
|
||||
crate::openhuman::context::DEFAULT_TOOL_RESULT_BUDGET_BYTES
|
||||
}
|
||||
|
||||
fn default_summarizer_payload_threshold_tokens() -> usize {
|
||||
500_000
|
||||
}
|
||||
|
||||
fn default_summarizer_max_payload_tokens() -> usize {
|
||||
2_000_000
|
||||
}
|
||||
|
||||
impl Default for ContextConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
@@ -123,6 +162,8 @@ impl Default for ContextConfig {
|
||||
reserve_output_tokens: default_reserve_output_tokens(),
|
||||
microcompact_keep_recent: default_microcompact_keep_recent(),
|
||||
tool_result_budget_bytes: default_tool_result_budget_bytes(),
|
||||
summarizer_payload_threshold_tokens: default_summarizer_payload_threshold_tokens(),
|
||||
summarizer_max_payload_tokens: default_summarizer_max_payload_tokens(),
|
||||
session_memory: SessionMemoryConfig::default(),
|
||||
summarizer_model: None,
|
||||
}
|
||||
|
||||
@@ -726,6 +726,7 @@ mod tests {
|
||||
disallowed_tools: vec![],
|
||||
skill_filter: None,
|
||||
category_filter: Some(ToolCategory::Skill),
|
||||
extra_tools: vec![],
|
||||
max_iterations: 8,
|
||||
timeout_secs: None,
|
||||
sandbox_mode: SandboxMode::None,
|
||||
@@ -908,6 +909,7 @@ mod tests {
|
||||
disallowed_tools: vec![],
|
||||
skill_filter: None,
|
||||
category_filter: None,
|
||||
extra_tools: vec![],
|
||||
max_iterations: 8,
|
||||
timeout_secs: None,
|
||||
sandbox_mode: SandboxMode::None,
|
||||
@@ -1075,6 +1077,7 @@ mod tests {
|
||||
disallowed_tools: vec![],
|
||||
skill_filter: None,
|
||||
category_filter: None,
|
||||
extra_tools: vec![],
|
||||
max_iterations: 2,
|
||||
timeout_secs: None,
|
||||
sandbox_mode: SandboxMode::None,
|
||||
@@ -1124,6 +1127,7 @@ mod tests {
|
||||
disallowed_tools: vec![],
|
||||
skill_filter: None,
|
||||
category_filter: None,
|
||||
extra_tools: vec![],
|
||||
max_iterations: 2,
|
||||
timeout_secs: None,
|
||||
sandbox_mode: SandboxMode::None,
|
||||
@@ -1181,6 +1185,7 @@ mod tests {
|
||||
disallowed_tools: vec![],
|
||||
skill_filter: None,
|
||||
category_filter: None,
|
||||
extra_tools: vec![],
|
||||
max_iterations: 2,
|
||||
timeout_secs: None,
|
||||
sandbox_mode: SandboxMode::None,
|
||||
|
||||
@@ -0,0 +1,497 @@
|
||||
use crate::openhuman::security::SecurityPolicy;
|
||||
use crate::openhuman::tools::traits::{Tool, ToolResult};
|
||||
use async_trait::async_trait;
|
||||
use serde_json::json;
|
||||
use std::sync::Arc;
|
||||
|
||||
/// Export structured data (JSON array of objects) as a CSV file to the workspace.
|
||||
pub struct CsvExportTool {
|
||||
security: Arc<SecurityPolicy>,
|
||||
}
|
||||
|
||||
impl CsvExportTool {
|
||||
pub fn new(security: Arc<SecurityPolicy>) -> Self {
|
||||
Self { security }
|
||||
}
|
||||
}
|
||||
|
||||
/// Escape a value for inclusion in a CSV cell. Wraps the value in
|
||||
/// double-quotes when it contains commas, quotes, or newlines. Embedded
|
||||
/// double-quotes are escaped by doubling them per RFC 4180.
|
||||
fn csv_escape(value: &str) -> String {
|
||||
if value.contains(',') || value.contains('"') || value.contains('\n') || value.contains('\r') {
|
||||
let escaped = value.replace('"', "\"\"");
|
||||
format!("\"{escaped}\"")
|
||||
} else {
|
||||
value.to_string()
|
||||
}
|
||||
}
|
||||
|
||||
/// Convert a `serde_json::Value` into a plain string suitable for a CSV
|
||||
/// cell. Objects and arrays are serialised as compact JSON; booleans and
|
||||
/// numbers use their natural representation; nulls become the empty
|
||||
/// string.
|
||||
fn value_to_cell(v: &serde_json::Value) -> String {
|
||||
match v {
|
||||
serde_json::Value::Null => String::new(),
|
||||
serde_json::Value::String(s) => s.clone(),
|
||||
serde_json::Value::Bool(b) => b.to_string(),
|
||||
serde_json::Value::Number(n) => n.to_string(),
|
||||
// Nested objects/arrays → compact JSON string
|
||||
other => other.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Collect column headers from a JSON array. If `columns` is provided,
|
||||
/// use those in order. Otherwise, collect all keys from the first object
|
||||
/// in the array (sorted alphabetically — serde_json uses BTreeMap by
|
||||
/// default). Callers who need a specific column order should pass the
|
||||
/// `columns` parameter.
|
||||
fn resolve_columns(items: &[serde_json::Value], columns: Option<&[String]>) -> Vec<String> {
|
||||
if let Some(cols) = columns {
|
||||
return cols.to_vec();
|
||||
}
|
||||
// Collect keys from the first object.
|
||||
if let Some(first) = items.first() {
|
||||
if let Some(obj) = first.as_object() {
|
||||
return obj.keys().cloned().collect();
|
||||
}
|
||||
}
|
||||
Vec::new()
|
||||
}
|
||||
|
||||
/// Render a JSON array of objects into a CSV string.
|
||||
fn render_csv(items: &[serde_json::Value], columns: &[String]) -> String {
|
||||
let mut buf = String::new();
|
||||
|
||||
// Header row
|
||||
let header: Vec<String> = columns.iter().map(|c| csv_escape(c)).collect();
|
||||
buf.push_str(&header.join(","));
|
||||
buf.push('\n');
|
||||
|
||||
// Data rows
|
||||
for item in items {
|
||||
let row: Vec<String> = columns
|
||||
.iter()
|
||||
.map(|col| {
|
||||
let cell_value = item.get(col).map(value_to_cell).unwrap_or_default();
|
||||
csv_escape(&cell_value)
|
||||
})
|
||||
.collect();
|
||||
buf.push_str(&row.join(","));
|
||||
buf.push('\n');
|
||||
}
|
||||
|
||||
buf
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Tool for CsvExportTool {
|
||||
fn name(&self) -> &str {
|
||||
"csv_export"
|
||||
}
|
||||
|
||||
fn description(&self) -> &str {
|
||||
"Export structured data (JSON array of objects) as a CSV file to the workspace. \
|
||||
Returns the file path. Use when the user wants raw tabular data from a tool \
|
||||
result that's too large to include inline."
|
||||
}
|
||||
|
||||
fn parameters_schema(&self) -> serde_json::Value {
|
||||
json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"data": {
|
||||
"type": "string",
|
||||
"description": "JSON string containing an array of objects to export. Each object becomes a row; keys become column headers."
|
||||
},
|
||||
"filename": {
|
||||
"type": "string",
|
||||
"description": "Output filename (without path). Will be written to workspace/exports/. Example: 'github-issues-2026-04-16.csv'"
|
||||
},
|
||||
"columns": {
|
||||
"type": "array",
|
||||
"items": { "type": "string" },
|
||||
"description": "Optional ordered list of column names to include. If omitted, all keys from the first object are used as headers."
|
||||
}
|
||||
},
|
||||
"required": ["data", "filename"]
|
||||
})
|
||||
}
|
||||
|
||||
fn permission_level(&self) -> crate::openhuman::tools::traits::PermissionLevel {
|
||||
crate::openhuman::tools::traits::PermissionLevel::Write
|
||||
}
|
||||
|
||||
async fn execute(&self, args: serde_json::Value) -> anyhow::Result<ToolResult> {
|
||||
let data_str = args
|
||||
.get("data")
|
||||
.and_then(|v| v.as_str())
|
||||
.ok_or_else(|| anyhow::anyhow!("Missing 'data' parameter"))?;
|
||||
|
||||
let filename = args
|
||||
.get("filename")
|
||||
.and_then(|v| v.as_str())
|
||||
.ok_or_else(|| anyhow::anyhow!("Missing 'filename' parameter"))?;
|
||||
|
||||
let columns: Option<Vec<String>> = args.get("columns").and_then(|v| {
|
||||
v.as_array().map(|arr| {
|
||||
arr.iter()
|
||||
.filter_map(|item| item.as_str().map(String::from))
|
||||
.collect()
|
||||
})
|
||||
});
|
||||
|
||||
// Security: check write permission
|
||||
if !self.security.can_act() {
|
||||
return Ok(ToolResult::error("Action blocked: autonomy is read-only"));
|
||||
}
|
||||
|
||||
if self.security.is_rate_limited() {
|
||||
return Ok(ToolResult::error(
|
||||
"Rate limit exceeded: too many actions in the last hour",
|
||||
));
|
||||
}
|
||||
|
||||
// Parse the JSON data
|
||||
let parsed: serde_json::Value = match serde_json::from_str(data_str) {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
return Ok(ToolResult::error(format!(
|
||||
"Failed to parse data as JSON: {e}"
|
||||
)));
|
||||
}
|
||||
};
|
||||
|
||||
let items = match parsed.as_array() {
|
||||
Some(arr) => arr,
|
||||
None => {
|
||||
return Ok(ToolResult::error(
|
||||
"Data must be a JSON array of objects, but got a non-array value",
|
||||
));
|
||||
}
|
||||
};
|
||||
|
||||
if items.is_empty() {
|
||||
return Ok(ToolResult::error("Data array is empty — nothing to export"));
|
||||
}
|
||||
|
||||
// Resolve columns and render CSV
|
||||
let cols = resolve_columns(items, columns.as_deref());
|
||||
let csv_content = render_csv(items, &cols);
|
||||
let csv_bytes = csv_content.len();
|
||||
|
||||
// Validate the relative path
|
||||
let relative_path = format!("exports/{filename}");
|
||||
if !self.security.is_path_allowed(&relative_path) {
|
||||
return Ok(ToolResult::error(format!(
|
||||
"Path not allowed by security policy: {relative_path}"
|
||||
)));
|
||||
}
|
||||
|
||||
let full_path = self.security.workspace_dir.join(&relative_path);
|
||||
|
||||
let Some(parent) = full_path.parent() else {
|
||||
return Ok(ToolResult::error("Invalid path: missing parent directory"));
|
||||
};
|
||||
|
||||
// Ensure exports/ directory exists
|
||||
tokio::fs::create_dir_all(parent).await?;
|
||||
|
||||
// Resolve parent AFTER creation to block symlink escapes.
|
||||
let resolved_parent = match tokio::fs::canonicalize(parent).await {
|
||||
Ok(p) => p,
|
||||
Err(e) => {
|
||||
return Ok(ToolResult::error(format!(
|
||||
"Failed to resolve file path: {e}"
|
||||
)));
|
||||
}
|
||||
};
|
||||
|
||||
if !self.security.is_resolved_path_allowed(&resolved_parent) {
|
||||
return Ok(ToolResult::error(format!(
|
||||
"Resolved path escapes workspace: {}",
|
||||
resolved_parent.display()
|
||||
)));
|
||||
}
|
||||
|
||||
let Some(file_name) = full_path.file_name() else {
|
||||
return Ok(ToolResult::error("Invalid path: missing file name"));
|
||||
};
|
||||
|
||||
let resolved_target = resolved_parent.join(file_name);
|
||||
|
||||
// If the target already exists and is a symlink, refuse to follow it
|
||||
if let Ok(meta) = tokio::fs::symlink_metadata(&resolved_target).await {
|
||||
if meta.file_type().is_symlink() {
|
||||
return Ok(ToolResult::error(format!(
|
||||
"Refusing to write through symlink: {}",
|
||||
resolved_target.display()
|
||||
)));
|
||||
}
|
||||
}
|
||||
|
||||
if !self.security.record_action() {
|
||||
return Ok(ToolResult::error(
|
||||
"Rate limit exceeded: action budget exhausted",
|
||||
));
|
||||
}
|
||||
|
||||
// Write the CSV file
|
||||
match tokio::fs::write(&resolved_target, &csv_content).await {
|
||||
Ok(()) => {
|
||||
let size_display = if csv_bytes >= 1024 * 1024 {
|
||||
format!("{:.1} MB", csv_bytes as f64 / (1024.0 * 1024.0))
|
||||
} else if csv_bytes >= 1024 {
|
||||
format!("{:.1} KB", csv_bytes as f64 / 1024.0)
|
||||
} else {
|
||||
format!("{csv_bytes} bytes")
|
||||
};
|
||||
|
||||
Ok(ToolResult::success(format!(
|
||||
"Exported {} rows to {relative_path} ({size_display})",
|
||||
items.len()
|
||||
)))
|
||||
}
|
||||
Err(e) => Ok(ToolResult::error(format!("Failed to write CSV file: {e}"))),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::openhuman::security::{AutonomyLevel, SecurityPolicy};
|
||||
|
||||
fn test_security(workspace: std::path::PathBuf) -> Arc<SecurityPolicy> {
|
||||
Arc::new(SecurityPolicy {
|
||||
autonomy: AutonomyLevel::Supervised,
|
||||
workspace_dir: workspace,
|
||||
..SecurityPolicy::default()
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn csv_export_name() {
|
||||
let tool = CsvExportTool::new(test_security(std::env::temp_dir()));
|
||||
assert_eq!(tool.name(), "csv_export");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn csv_export_schema_has_required_fields() {
|
||||
let tool = CsvExportTool::new(test_security(std::env::temp_dir()));
|
||||
let schema = tool.parameters_schema();
|
||||
assert!(schema["properties"]["data"].is_object());
|
||||
assert!(schema["properties"]["filename"].is_object());
|
||||
assert!(schema["properties"]["columns"].is_object());
|
||||
let required = schema["required"].as_array().unwrap();
|
||||
assert!(required.contains(&json!("data")));
|
||||
assert!(required.contains(&json!("filename")));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn csv_export_formats_simple_array() {
|
||||
let dir = std::env::temp_dir().join("openhuman_test_csv_export_simple");
|
||||
let _ = tokio::fs::remove_dir_all(&dir).await;
|
||||
tokio::fs::create_dir_all(&dir).await.unwrap();
|
||||
|
||||
let tool = CsvExportTool::new(test_security(dir.clone()));
|
||||
let data = serde_json::to_string(&json!([
|
||||
{"name": "Alice", "age": 30, "city": "NYC"},
|
||||
{"name": "Bob", "age": 25, "city": "LA"},
|
||||
{"name": "Carol", "age": 35, "city": "Chicago"}
|
||||
]))
|
||||
.unwrap();
|
||||
|
||||
let result = tool
|
||||
.execute(json!({
|
||||
"data": data,
|
||||
"filename": "people.csv"
|
||||
}))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert!(!result.is_error, "unexpected error: {}", result.output());
|
||||
assert!(result.output().contains("3 rows"));
|
||||
|
||||
let content = tokio::fs::read_to_string(dir.join("exports/people.csv"))
|
||||
.await
|
||||
.unwrap();
|
||||
let lines: Vec<&str> = content.trim().lines().collect();
|
||||
assert_eq!(lines.len(), 4, "header + 3 data rows");
|
||||
|
||||
// Header should contain the keys from the first object
|
||||
let header = lines[0];
|
||||
assert!(header.contains("name"));
|
||||
assert!(header.contains("age"));
|
||||
assert!(header.contains("city"));
|
||||
|
||||
// Data rows should contain values
|
||||
assert!(lines[1].contains("Alice"));
|
||||
assert!(lines[2].contains("Bob"));
|
||||
assert!(lines[3].contains("Carol"));
|
||||
|
||||
let _ = tokio::fs::remove_dir_all(&dir).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn csv_export_handles_missing_keys() {
|
||||
let dir = std::env::temp_dir().join("openhuman_test_csv_export_missing_keys");
|
||||
let _ = tokio::fs::remove_dir_all(&dir).await;
|
||||
tokio::fs::create_dir_all(&dir).await.unwrap();
|
||||
|
||||
let tool = CsvExportTool::new(test_security(dir.clone()));
|
||||
let data = serde_json::to_string(&json!([
|
||||
{"name": "Alice", "age": 30, "city": "NYC"},
|
||||
{"name": "Bob"},
|
||||
{"name": "Carol", "city": "Chicago"}
|
||||
]))
|
||||
.unwrap();
|
||||
|
||||
let result = tool
|
||||
.execute(json!({
|
||||
"data": data,
|
||||
"filename": "sparse.csv",
|
||||
"columns": ["name", "age", "city"]
|
||||
}))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert!(!result.is_error, "unexpected error: {}", result.output());
|
||||
|
||||
let content = tokio::fs::read_to_string(dir.join("exports/sparse.csv"))
|
||||
.await
|
||||
.unwrap();
|
||||
let lines: Vec<&str> = content.trim().lines().collect();
|
||||
assert_eq!(lines.len(), 4);
|
||||
|
||||
// Bob's row should have empty cells for age and city
|
||||
let bob_row = lines[2];
|
||||
let bob_cells: Vec<&str> = bob_row.split(',').collect();
|
||||
assert_eq!(bob_cells.len(), 3, "Bob row should have 3 cells");
|
||||
assert_eq!(bob_cells[0], "Bob");
|
||||
assert_eq!(bob_cells[1], "", "missing age should be empty");
|
||||
assert_eq!(bob_cells[2], "", "missing city should be empty");
|
||||
|
||||
let _ = tokio::fs::remove_dir_all(&dir).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn csv_export_respects_column_order() {
|
||||
let dir = std::env::temp_dir().join("openhuman_test_csv_export_column_order");
|
||||
let _ = tokio::fs::remove_dir_all(&dir).await;
|
||||
tokio::fs::create_dir_all(&dir).await.unwrap();
|
||||
|
||||
let tool = CsvExportTool::new(test_security(dir.clone()));
|
||||
let data = serde_json::to_string(&json!([
|
||||
{"name": "Alice", "age": 30, "city": "NYC"},
|
||||
{"name": "Bob", "age": 25, "city": "LA"}
|
||||
]))
|
||||
.unwrap();
|
||||
|
||||
let result = tool
|
||||
.execute(json!({
|
||||
"data": data,
|
||||
"filename": "ordered.csv",
|
||||
"columns": ["city", "name", "age"]
|
||||
}))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert!(!result.is_error, "unexpected error: {}", result.output());
|
||||
|
||||
let content = tokio::fs::read_to_string(dir.join("exports/ordered.csv"))
|
||||
.await
|
||||
.unwrap();
|
||||
let lines: Vec<&str> = content.trim().lines().collect();
|
||||
assert_eq!(
|
||||
lines[0], "city,name,age",
|
||||
"header must follow requested column order"
|
||||
);
|
||||
assert_eq!(lines[1], "NYC,Alice,30");
|
||||
assert_eq!(lines[2], "LA,Bob,25");
|
||||
|
||||
let _ = tokio::fs::remove_dir_all(&dir).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn csv_export_rejects_non_array_input() {
|
||||
let dir = std::env::temp_dir().join("openhuman_test_csv_export_non_array");
|
||||
let _ = tokio::fs::remove_dir_all(&dir).await;
|
||||
tokio::fs::create_dir_all(&dir).await.unwrap();
|
||||
|
||||
let tool = CsvExportTool::new(test_security(dir.clone()));
|
||||
let data = serde_json::to_string(&json!({"not": "an array"})).unwrap();
|
||||
|
||||
let result = tool
|
||||
.execute(json!({
|
||||
"data": data,
|
||||
"filename": "bad.csv"
|
||||
}))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert!(result.is_error);
|
||||
assert!(
|
||||
result.output().contains("non-array"),
|
||||
"error should mention non-array, got: {}",
|
||||
result.output()
|
||||
);
|
||||
|
||||
let _ = tokio::fs::remove_dir_all(&dir).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn csv_export_handles_nested_values() {
|
||||
let dir = std::env::temp_dir().join("openhuman_test_csv_export_nested");
|
||||
let _ = tokio::fs::remove_dir_all(&dir).await;
|
||||
tokio::fs::create_dir_all(&dir).await.unwrap();
|
||||
|
||||
let tool = CsvExportTool::new(test_security(dir.clone()));
|
||||
let data = serde_json::to_string(&json!([
|
||||
{
|
||||
"name": "Alice",
|
||||
"tags": ["admin", "dev"],
|
||||
"meta": {"role": "lead", "level": 5}
|
||||
},
|
||||
{
|
||||
"name": "Bob",
|
||||
"tags": [],
|
||||
"meta": null
|
||||
}
|
||||
]))
|
||||
.unwrap();
|
||||
|
||||
let result = tool
|
||||
.execute(json!({
|
||||
"data": data,
|
||||
"filename": "nested.csv",
|
||||
"columns": ["name", "tags", "meta"]
|
||||
}))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert!(!result.is_error, "unexpected error: {}", result.output());
|
||||
|
||||
let content = tokio::fs::read_to_string(dir.join("exports/nested.csv"))
|
||||
.await
|
||||
.unwrap();
|
||||
let lines: Vec<&str> = content.trim().lines().collect();
|
||||
assert_eq!(lines.len(), 3, "header + 2 data rows");
|
||||
|
||||
// Alice's tags should be serialized as a JSON string (in quotes because it contains commas)
|
||||
let alice_row = lines[1];
|
||||
assert!(alice_row.contains("Alice"));
|
||||
// The JSON array should be serialized as a string and quoted
|
||||
assert!(
|
||||
alice_row.contains(r#"[""admin"",""dev""]"#),
|
||||
"nested arrays should be JSON-serialized in CSV: {alice_row}"
|
||||
);
|
||||
|
||||
// Bob's meta is null → empty cell
|
||||
let bob_row = lines[2];
|
||||
assert!(bob_row.contains("Bob"));
|
||||
|
||||
let _ = tokio::fs::remove_dir_all(&dir).await;
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
mod csv_export;
|
||||
mod file_read;
|
||||
mod file_write;
|
||||
mod git_operations;
|
||||
@@ -6,6 +7,7 @@ mod run_linter;
|
||||
mod run_tests;
|
||||
mod update_memory_md;
|
||||
|
||||
pub use csv_export::CsvExportTool;
|
||||
pub use file_read::FileReadTool;
|
||||
pub use file_write::FileWriteTool;
|
||||
pub use git_operations::GitOperationsTool;
|
||||
|
||||
@@ -75,6 +75,7 @@ pub fn all_tools_with_runtime(
|
||||
Box::new(ShellTool::new(security.clone(), runtime)),
|
||||
Box::new(FileReadTool::new(security.clone())),
|
||||
Box::new(FileWriteTool::new(security.clone())),
|
||||
Box::new(CsvExportTool::new(security.clone())),
|
||||
// Sub-agent dispatch — lets the parent agent delegate focused
|
||||
// sub-tasks (research, code execution, API specialists, …) by
|
||||
// calling `spawn_subagent { agent_id, prompt, … }`. The runner
|
||||
|
||||
@@ -68,6 +68,20 @@ pub fn collect_orchestrator_tools(
|
||||
for entry in &definition.subagents {
|
||||
match entry {
|
||||
SubagentEntry::AgentId(agent_id) => {
|
||||
// Runtime-only sub-agents — the LLM must never see a
|
||||
// `delegate_*` tool for these because they're dispatched
|
||||
// directly by the runtime, not by an explicit LLM tool
|
||||
// call. Issue #574 introduced `summarizer` as the first
|
||||
// such sub-agent; future runtime-only agents should
|
||||
// join this filter.
|
||||
if agent_id == "summarizer" {
|
||||
log::debug!(
|
||||
"[orchestrator_tools] skipping runtime-only sub-agent '{}' \
|
||||
(no delegation tool synthesised)",
|
||||
agent_id
|
||||
);
|
||||
continue;
|
||||
}
|
||||
let Some(target) = registry.get(agent_id) else {
|
||||
log::warn!(
|
||||
"[orchestrator_tools] subagent '{}' referenced by '{}' is not in the registry — skipping",
|
||||
@@ -189,6 +203,7 @@ mod tests {
|
||||
disallowed_tools: vec![],
|
||||
skill_filter: None,
|
||||
category_filter: None,
|
||||
extra_tools: vec![],
|
||||
max_iterations: 8,
|
||||
timeout_secs: None,
|
||||
sandbox_mode: SandboxMode::None,
|
||||
|
||||
Reference in New Issue
Block a user