Merge pull request #141 from living-ip/codex/leo-db-first-postdeploy-budget-repair-20260714

Enforce Leo reply budgets before delivery

hermes-agent/leoclean-bin/kb_tool.py

@@ -525,8 +525,8 @@ def operational_contracts(
     contracts: list[dict[str, Any]] = [
         {
             "id": "reply_budget",
-            "target_words": 150,
+            "target_words": 170,
-            "hard_max_words": 200,
+            "hard_max_words": 220,
             "shape": (
                 "no intro; at most three compact bullets covering mapping, review/apply, and receipt/limitation; "
                 "omission is better than exceeding the budget"

hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py

@@ -18,6 +18,9 @@ from typing import Any
 DEFAULT_TIMEOUT_SECONDS = 10
 MAX_QUERY_CHARS = 16_000
 MAX_PENDING_SNAPSHOTS = 128
+WORD_RE = re.compile(r"\b\w+(?:[-']\w+)*\b")
+LIST_PREFIX_RE = re.compile(r"^(\s*(?:[-*]|\d+[.)])\s+)(.*)$")
+SENTENCE_BREAK_RE = re.compile(r"(?<=[.!?])\s+")
 COMPILED_CONTRACT_IDS = frozenset(
     {
         "mixed_packet_composition",
@@ -151,7 +154,8 @@ def _format_context(contracts: list[dict[str, Any]]) -> str:
         '<leo_current_runtime_contracts source="live teleo-kb context" status="ok">\n'
         "This trusted, read-only block was generated automatically for the current question. It is the "
         "binding contract for current schema and runtime capability. Base the answer on it, do not invent "
-        "fields or capabilities, respect its reply budget, and distinguish canonical rows from staging. "
+        "fields or capabilities, draft at or below target_words, never exceed hard_max_words, and distinguish "
+        "canonical rows from staging. The hard limit is enforced before delivery. "
         "Do not claim that you called a tool; this context was injected before reasoning.\n"
         f"{payload}\n"
         "</leo_current_runtime_contracts>"
@@ -278,7 +282,69 @@ def build_database_context(
 
 
 def _word_count(value: str) -> int:
-    return len(re.findall(r"\b\w+(?:[-']\w+)*\b", value))
+    return len(WORD_RE.findall(value))
+
+
+def _truncate_to_word_budget(value: str, hard_max: int) -> str:
+    matches = list(WORD_RE.finditer(value))
+    if len(matches) <= hard_max:
+        return value.strip()
+    truncated = value[: matches[hard_max - 1].end()].rstrip()
+    return truncated if truncated.endswith((".", "!", "?", ":", ";")) else truncated + "..."
+
+
+def _budget_line_parts(line: str) -> tuple[str, list[str]]:
+    match = LIST_PREFIX_RE.match(line)
+    prefix, body = (match.group(1), match.group(2)) if match else ("", line.strip())
+    sentences = [part.strip() for part in SENTENCE_BREAK_RE.split(body) if part.strip()]
+    return prefix, sentences or [body]
+
+
+def _compact_response_to_budget(response: str, hard_max: int) -> str:
+    """Keep every paragraph/list item's lead sentence, then add detail while it fits."""
+
+    if hard_max < 1 or _word_count(response) <= hard_max:
+        return response.strip()
+
+    lines = response.splitlines()
+    records: dict[int, dict[str, Any]] = {}
+    for index, line in enumerate(lines):
+        if not line.strip():
+            continue
+        prefix, sentences = _budget_line_parts(line)
+        records[index] = {"prefix": prefix, "sentences": sentences, "selected": 1}
+
+    def render() -> str:
+        rendered: list[str] = []
+        for index, _line in enumerate(lines):
+            record = records.get(index)
+            if record is None:
+                rendered.append("")
+                continue
+            selected = int(record["selected"])
+            rendered.append(str(record["prefix"]) + " ".join(record["sentences"][:selected]))
+        return re.sub(r"\n{3,}", "\n\n", "\n".join(rendered)).strip()
+
+    compacted = render()
+    if _word_count(compacted) > hard_max:
+        return _truncate_to_word_budget(compacted, hard_max)
+
+    while True:
+        progress = False
+        for record in records.values():
+            selected = int(record["selected"])
+            if selected >= len(record["sentences"]):
+                continue
+            record["selected"] = selected + 1
+            candidate = render()
+            if _word_count(candidate) <= hard_max:
+                compacted = candidate
+                progress = True
+            else:
+                record["selected"] = selected
+        if not progress:
+            break
+    return compacted
 
 
 def _reply_hard_max(contracts: list[dict[str, Any]]) -> int | None:
@@ -625,21 +691,33 @@ def _post_llm_call(**kwargs: Any) -> dict[str, str] | None:
     compiled_valid = bool(isinstance(compiled_response, str) and compiled_response.strip() and not compiled_issues)
     fail_closed = bool(compiled_required and not compiled_valid)
     enforce_compiled = bool(compiled_required and compiled_valid)
-    transformed = bool(fail_closed or (enforce_compiled and str(compiled_response) != response))
     if fail_closed:
         delivered = DATABASE_UNAVAILABLE_RESPONSE
     elif enforce_compiled or (issues and compiled_valid):
         delivered = str(compiled_response)
     else:
         delivered = response
+    hard_max = _reply_hard_max(contracts)
+    budget_compacted = bool(
+        not fail_closed
+        and delivered == response
+        and hard_max is not None
+        and "reply_budget_exceeded" in issues
+    )
+    if budget_compacted:
+        delivered = _compact_response_to_budget(delivered, hard_max)
+    delivered_issues = response_contract_issues(delivered, contracts)
+    transformed = delivered != response
     _trace(
         base_record
         | {
-            "status": "error" if fail_closed else "ok" if not issues or delivered != response else "error",
+            "status": "error" if fail_closed or delivered_issues else "ok",
             "contract_ids": sorted(contract_ids),
             "issues": issues,
+            "delivered_issues": delivered_issues,
             "compiled_response_issues": compiled_issues,
             "transformed": transformed,
+            "budget_compacted": budget_compacted,
             "compiled_response_enforced": enforce_compiled,
             "fail_closed": fail_closed,
             "delivered_response_sha256": hashlib.sha256(delivered.encode("utf-8")).hexdigest(),

tests/test_hermes_leoclean_db_context_plugin.py

@@ -154,6 +154,49 @@ def test_plugin_keeps_contract_valid_draft(tmp_path: Path) -> None:
     )
 
 
+def test_plugin_compacts_generic_over_budget_reply_without_a_query_template(tmp_path: Path, monkeypatch) -> None:
+    module = load_plugin()
+    trace = tmp_path / "trace.jsonl"
+    monkeypatch.setenv("LEO_DB_CONTEXT_TRACE_PATH", str(trace))
+    query = "Challenge this claim and propose a better one."
+    contracts = [{"id": "reply_budget", "hard_max_words": 24}]
+    module._store_snapshot(
+        "budget-session",
+        {
+            "status": "ok",
+            "query_sha256": module.hashlib.sha256(query.encode("utf-8")).hexdigest(),
+            "contracts": contracts,
+            "compiled_response": None,
+            "requires_database_truth": False,
+            "context": "fixture",
+        },
+    )
+    draft = (
+        "Evidence is weak. This sentence contains expendable evidence detail for the draft.\n\n"
+        "- The claim is too broad. This sentence is extra explanation.\n"
+        "1. Propose a narrower claim. This sentence is extra proposal detail.\n\n"
+        "What should we challenge first?"
+    )
+
+    result = module._post_llm_call(
+        session_id="budget-session", user_message=query, assistant_response=draft
+    )
+
+    assert result is not None
+    delivered = result["assistant_response"]
+    assert module._word_count(delivered) <= 24
+    assert "Evidence is weak." in delivered
+    assert "- The claim is too broad." in delivered
+    assert "1. Propose a narrower claim." in delivered
+    assert "What should we challenge first?" in delivered
+    record = json.loads(trace.read_text(encoding="utf-8"))
+    assert record["status"] == "ok"
+    assert record["budget_compacted"] is True
+    assert record["issues"] == ["reply_budget_exceeded"]
+    assert record["delivered_issues"] == []
+    assert record["transformed"] is True
+
+
 def test_source_validator_accepts_explicit_schema_and_canonical_write_prohibitions() -> None:
     module = load_plugin()
     response = (

tests/test_hermes_leoclean_kb_bridge_source.py

@@ -392,8 +392,8 @@ def test_vps_bridge_operational_contracts_are_query_specific_and_schema_exact()
     module = _load_module(BRIDGE_DIR / "kb_tool.py")
 
     reply_budget = module.operational_contracts("How should Leo answer?")[0]
-    assert reply_budget["target_words"] == 150
+    assert reply_budget["target_words"] == 170
-    assert reply_budget["hard_max_words"] == 200
+    assert reply_budget["hard_max_words"] == 220
     assert "omission is better" in reply_budget["shape"]
 
     source_contracts = {