140 lines
5.1 KiB
Python
140 lines
5.1 KiB
Python
"""Database-driven runtime context for the leoclean Hermes profile."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
|
|
DEFAULT_TIMEOUT_SECONDS = 10
|
|
MAX_QUERY_CHARS = 16_000
|
|
|
|
|
|
def _trace(record: dict[str, Any]) -> None:
|
|
"""Write test-only proof without retaining the operator's message."""
|
|
|
|
raw_path = os.getenv("LEO_DB_CONTEXT_TRACE_PATH", "").strip()
|
|
if not raw_path:
|
|
return
|
|
path = Path(raw_path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("a", encoding="utf-8") as handle:
|
|
handle.write(json.dumps(record, sort_keys=True) + "\n")
|
|
try:
|
|
path.chmod(0o600)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def _failure_context(reason: str) -> str:
|
|
return (
|
|
"<leo_current_runtime_contracts source=\"teleo-kb\" status=\"unavailable\">\n"
|
|
"The read-only current database contract lookup failed. Do not infer current schema, row state, "
|
|
"proposal applyability, or source-ingestion capability from memory. State that live readback is "
|
|
f"unavailable and use the teleo-kb bridge before making a database claim. Failure: {reason}\n"
|
|
"</leo_current_runtime_contracts>"
|
|
)
|
|
|
|
|
|
def _format_context(contracts: list[dict[str, Any]]) -> str:
|
|
payload = json.dumps(contracts, sort_keys=True, separators=(",", ":"))
|
|
return (
|
|
"<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. "
|
|
"Do not claim that you called a tool; this context was injected before reasoning.\n"
|
|
f"{payload}\n"
|
|
"</leo_current_runtime_contracts>"
|
|
)
|
|
|
|
|
|
def build_database_context(
|
|
user_message: str,
|
|
*,
|
|
hermes_home: Path | None = None,
|
|
runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
|
|
) -> str:
|
|
"""Return current operational contracts for one exact operator question."""
|
|
|
|
query = str(user_message or "")[:MAX_QUERY_CHARS]
|
|
home = hermes_home or Path(os.getenv("HERMES_HOME", str(Path.home() / ".hermes")))
|
|
kb_tool = home / "bin" / "kb_tool.py"
|
|
query_sha256 = hashlib.sha256(query.encode("utf-8")).hexdigest()
|
|
base_record: dict[str, Any] = {
|
|
"generated_at_utc": datetime.now(timezone.utc).isoformat(),
|
|
"query_sha256": query_sha256,
|
|
"source": "kb_tool.py --local context",
|
|
}
|
|
|
|
if not kb_tool.is_file():
|
|
record = base_record | {"status": "error", "error": "kb_tool_missing", "injected": True}
|
|
_trace(record)
|
|
return _failure_context("kb_tool_missing")
|
|
|
|
timeout = int(os.getenv("LEO_DB_CONTEXT_TIMEOUT_SECONDS", str(DEFAULT_TIMEOUT_SECONDS)))
|
|
command = [
|
|
sys.executable,
|
|
str(kb_tool),
|
|
"--local",
|
|
"context",
|
|
query,
|
|
"--limit",
|
|
"0",
|
|
"--context-limit",
|
|
"0",
|
|
"--format",
|
|
"json",
|
|
]
|
|
try:
|
|
completed = runner(command, capture_output=True, text=True, timeout=timeout, check=False)
|
|
except subprocess.TimeoutExpired:
|
|
record = base_record | {"status": "error", "error": "timeout", "injected": True}
|
|
_trace(record)
|
|
return _failure_context("timeout")
|
|
except OSError as exc:
|
|
record = base_record | {"status": "error", "error": type(exc).__name__, "injected": True}
|
|
_trace(record)
|
|
return _failure_context(type(exc).__name__)
|
|
|
|
if completed.returncode != 0:
|
|
record = base_record | {
|
|
"status": "error",
|
|
"error": f"kb_tool_exit_{completed.returncode}",
|
|
"injected": True,
|
|
}
|
|
_trace(record)
|
|
return _failure_context(f"kb_tool_exit_{completed.returncode}")
|
|
|
|
try:
|
|
payload = json.loads(completed.stdout)
|
|
contracts = payload.get("operational_contracts")
|
|
if not isinstance(contracts, list) or not all(isinstance(item, dict) for item in contracts):
|
|
raise ValueError("operational_contracts_missing")
|
|
except (json.JSONDecodeError, TypeError, ValueError) as exc:
|
|
record = base_record | {"status": "error", "error": str(exc), "injected": True}
|
|
_trace(record)
|
|
return _failure_context(str(exc))
|
|
|
|
contract_json = json.dumps(contracts, sort_keys=True, separators=(",", ":"))
|
|
record = base_record | {
|
|
"status": "ok",
|
|
"injected": True,
|
|
"contract_ids": [str(item.get("id") or "") for item in contracts],
|
|
"contract_sha256": hashlib.sha256(contract_json.encode("utf-8")).hexdigest(),
|
|
}
|
|
_trace(record)
|
|
return _format_context(contracts)
|
|
|
|
|
|
def _pre_llm_call(**kwargs: Any) -> dict[str, str]:
|
|
return {"context": build_database_context(str(kwargs.get("user_message") or ""))}
|
|
|
|
|
|
def register(ctx: Any) -> None:
|
|
ctx.register_hook("pre_llm_call", _pre_llm_call)
|