"""Truncate delegate summaries before injecting into parent context.""" from __future__ import annotations from typing import Any from butler.utilities.env_parse import float_env, int_env from butler.defaults.env_defaults import ( DELEGATE_SUMMARY_MAX_CHARS, DELEGATE_SUMMARY_RESERVE_RATIO, ) def delegate_summary_max_chars() -> int: return int(int_env("BUTLER_DELEGATE_SUMMARY_MAX_CHARS", DELEGATE_SUMMARY_MAX_CHARS, min=500, max=50000)) def delegate_summary_reserve_ratio() -> float: return float(float_env("BUTLER_DELEGATE_SUMMARY_RESERVE_RATIO", DELEGATE_SUMMARY_RESERVE_RATIO, min=0.0, max=0.5)) def truncate_delegate_summary(text: str, *, max_chars: int | None = None) -> str: """Keep structured prefix lines when truncating long delegate prose.""" cap = max_chars if max_chars is not None else delegate_summary_max_chars() body = (text or "").strip() if len(body) <= cap: return body head = body[: max(200, cap // 3)] tail_budget = cap - len(head) - 80 if tail_budget < 100: return body[:cap] + "\n…[truncated]" return head + "\n…\n" + body[-tail_budget:] + "\n…[truncated]" def budget_delegate_payload(payload: dict[str, Any]) -> dict[str, Any]: """Apply headroom budget to delegate tool result payload.""" out = dict(payload) summary = str(out.get("summary") or "") if summary: out["summary"] = truncate_delegate_summary(summary) if len(summary) > len(out["summary"]): out["summary_truncated"] = True return out __all__ = [ "budget_delegate_payload", "delegate_summary_max_chars", "delegate_summary_reserve_ratio", "truncate_delegate_summary", ]