teleo/teleo-infrastructure
8
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions 1

Enforce Leo reply budgets before delivery

Browse source
This commit is contained in:
twentyOne2x 2026-07-14 09:46:56 +02:00
parent cf22e13d5e
commit ad70bd4876
4 changed files with 129 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
hermes-agent/leoclean-bin/kb_tool.py
View file

@ -525,8 +525,8 @@ def operational_contracts(
contracts: list[dict[str, Any]] = [
{
"id": "reply_budget",
"target_words": 150,
"hard_max_words": 200,
"target_words": 170,
"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"

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

@ -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(),

43
tests/test_hermes_leoclean_db_context_plugin.py
View file

@ -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 = (

4
tests/test_hermes_leoclean_kb_bridge_source.py
View file

@ -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["hard_max_words"] == 200
assert reply_budget["target_words"] == 170
assert reply_budget["hard_max_words"] == 220
assert "omission is better" in reply_budget["shape"]
source_contracts = {