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

Inject live database contracts into Leo turns

Browse source
This commit is contained in:
twentyOne2x 2026-07-13 12:37:08 +02:00
parent d9d799d725
commit 5bc78a2a5a
12 changed files with 569 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
deploy/auto-deploy.sh
View file

@ -18,6 +18,7 @@ DIAGNOSTICS_DIR="/opt/teleo-eval/diagnostics"
AGENT_STATE_DIR="/opt/teleo-eval/ops/agent-state"
LEOCLEAN_BIN_DIR="/home/teleo/.hermes/profiles/leoclean/bin"
LEOCLEAN_SKILLS_DIR="/home/teleo/.hermes/profiles/leoclean/skills"
LEOCLEAN_PLUGINS_DIR="/home/teleo/.hermes/profiles/leoclean/plugins"
SYSTEMD_DIR="/etc/systemd/system"
STAMP_FILE="/opt/teleo-eval/.last-deploy-sha"
LOG_TAG="auto-deploy"
@ -68,7 +69,7 @@ fi
# Syntax check all Python files before copying
ERRORS=0
for f in lib/*.py *.py diagnostics/*.py telegram/*.py tests/*.py; do
for f in lib/*.py *.py diagnostics/*.py telegram/*.py tests/*.py hermes-agent/leoclean-plugins/vps/*/*.py; do
[ -f "$f" ] || continue
if ! python3 -c "import ast, sys; ast.parse(open(sys.argv[1]).read())" "$f" 2>&1; then
log "SYNTAX ERROR: $f"
@ -103,12 +104,16 @@ if [ -d hermes-agent/leoclean-skills/vps ]; then
mkdir -p "$LEOCLEAN_SKILLS_DIR"
rsync "${RSYNC_OPTS[@]}" hermes-agent/leoclean-skills/vps/ "$LEOCLEAN_SKILLS_DIR/"
fi
if [ -d hermes-agent/leoclean-plugins/vps ]; then
mkdir -p "$LEOCLEAN_PLUGINS_DIR"
rsync "${RSYNC_OPTS[@]}" hermes-agent/leoclean-plugins/vps/ "$LEOCLEAN_PLUGINS_DIR/"
fi
[ -f research/research-session.sh ] && rsync "${RSYNC_OPTS[@]}" research/research-session.sh /opt/teleo-eval/research-session.sh
# Safety net: ensure synced .sh files are executable after rsync.
# Keep this bounded to deploy-owned paths: /opt/teleo-eval also contains
# backups and generated state that may be unreadable by the deploy user.
for dir in "$PIPELINE_DIR" "$TELEGRAM_DIR" "$DIAGNOSTICS_DIR" "$AGENT_STATE_DIR" "$LEOCLEAN_BIN_DIR" "$LEOCLEAN_SKILLS_DIR"; do
for dir in "$PIPELINE_DIR" "$TELEGRAM_DIR" "$DIAGNOSTICS_DIR" "$AGENT_STATE_DIR" "$LEOCLEAN_BIN_DIR" "$LEOCLEAN_SKILLS_DIR" "$LEOCLEAN_PLUGINS_DIR"; do
[ -d "$dir" ] || continue
find "$dir" -maxdepth 3 -name '*.sh' -not -perm -u+x -exec chmod +x {} +
done
@ -164,7 +169,7 @@ if [ "$OLD_SHA" != "none" ]; then
if git diff --name-only "$OLD_SHA" "$NEW_SHA" -- diagnostics/ 2>/dev/null | grep -q '\.py$'; then
add_restart teleo-diagnostics
fi
if git diff --name-only "$OLD_SHA" "$NEW_SHA" -- hermes-agent/leoclean-bin/ hermes-agent/leoclean-skills/vps/ 2>/dev/null \
if git diff --name-only "$OLD_SHA" "$NEW_SHA" -- hermes-agent/leoclean-bin/ hermes-agent/leoclean-skills/vps/ hermes-agent/leoclean-plugins/vps/ 2>/dev/null \
| bash deploy/leoclean-gateway-restart-required.sh; then
add_restart_if_unit_active leoclean-gateway
fi

7
deploy/deploy.sh
View file

@ -12,6 +12,7 @@ VPS_DIAGNOSTICS="/opt/teleo-eval/diagnostics"
VPS_AGENT_STATE="/opt/teleo-eval/ops/agent-state"
VPS_LEOCLEAN_BIN="/home/teleo/.hermes/profiles/leoclean/bin"
VPS_LEOCLEAN_SKILLS="/home/teleo/.hermes/profiles/leoclean/skills"
VPS_LEOCLEAN_PLUGINS="/home/teleo/.hermes/profiles/leoclean/plugins"
VPS_SYSTEMD="/etc/systemd/system"
REPO_ROOT="$(cd "$(dirname "$0")/.." && pwd)"
@ -45,7 +46,7 @@ echo ""
# Syntax check all Python files before deploying
echo "=== Pre-deploy syntax check ==="
ERRORS=0
for f in "$REPO_ROOT/lib/"*.py "$REPO_ROOT/"*.py "$REPO_ROOT/diagnostics/"*.py "$REPO_ROOT/telegram/"*.py; do
for f in "$REPO_ROOT/lib/"*.py "$REPO_ROOT/"*.py "$REPO_ROOT/diagnostics/"*.py "$REPO_ROOT/telegram/"*.py "$REPO_ROOT/hermes-agent/leoclean-plugins/vps/"*/*.py; do
[ -f "$f" ] || continue
if ! python3 -c "import ast, sys; ast.parse(open(sys.argv[1]).read())" "$f" 2>/dev/null; then
echo "SYNTAX ERROR: $f"
@ -101,6 +102,10 @@ echo "=== Leoclean skills ==="
rsync "${RSYNC_OPTS[@]}" "$REPO_ROOT/hermes-agent/leoclean-skills/vps/" "$VPS_HOST:$VPS_LEOCLEAN_SKILLS/"
echo ""
echo "=== Leoclean plugins ==="
rsync "${RSYNC_OPTS[@]}" "$REPO_ROOT/hermes-agent/leoclean-plugins/vps/" "$VPS_HOST:$VPS_LEOCLEAN_PLUGINS/"
echo ""
echo "=== Research session ==="
rsync "${RSYNC_OPTS[@]}" "$REPO_ROOT/research/research-session.sh" "$VPS_HOST:/opt/teleo-eval/research-session.sh"
echo ""

3
deploy/leoclean-gateway-restart-required.sh
View file

@ -14,6 +14,9 @@ while IFS= read -r path || [ -n "$path" ]; do
hermes-agent/leoclean-skills/vps/*)
restart_required=0
;;
hermes-agent/leoclean-plugins/vps/*)
restart_required=0
;;
esac
done

156
docs/reports/leo-working-state-20260709/telegram-handler-m3taversal-oos-targeted-post-deterministic-context-contract-20260713-score.json Normal file
View file

@ -0,0 +1,156 @@
{
"db_counts_changed": false,
"generated_at_utc": "2026-07-13T10:28:02.926071+00:00",
"mode": "working_leo_m3taversal_targeted_deterministic_context_contract_score",
"posted_to_telegram": false,
"production_db_apply_ran": false,
"safe_pass": false,
"score": {
"expected_prompt_count": 2,
"failures": [
{
"behavioral_rule_schema_issues": [],
"broad_semantic_issues": [],
"concepts": {
"bounded_intake_tier": false,
"real_source_identity": false,
"staged_review_apply": true,
"staging_without_apply_authorization": false
},
"current_schema_overclaims": [
"sources_unshipped_fields"
],
"custom_signals": {},
"dimension": "autonomous_source_intake_boundary",
"invalid_count_invariant_detected": false,
"legacy_pass": false,
"legacy_signals": {
"artifact": true,
"authorization": true,
"no_overclaim": false,
"staging_or_review": false
},
"max_response_words": 220,
"overclaim_detected": false,
"pass": false,
"prompt_id": "OOS-14",
"proposal_readiness_issues": [],
"response_issue_detected": false,
"response_too_long": true,
"source_evidence_semantic_issues": [],
"source_intake_issues": [],
"word_count": 372
},
{
"behavioral_rule_schema_issues": [],
"broad_semantic_issues": [],
"concepts": {
"behavioral_rule_storage": false,
"heterogeneous_types": true,
"receipt": false,
"reviewed_policy_apply": false,
"staged_review_apply": true
},
"current_schema_overclaims": [],
"custom_signals": {},
"dimension": "database_composition_judgment",
"invalid_count_invariant_detected": false,
"legacy_pass": false,
"legacy_signals": {
"canonical_db": true,
"caveat_retention": false,
"next_action": true,
"no_overclaim": false,
"staging_or_review": false
},
"max_response_words": 220,
"overclaim_detected": false,
"pass": false,
"prompt_id": "OOS-06",
"proposal_readiness_issues": [],
"response_issue_detected": false,
"response_too_long": true,
"source_evidence_semantic_issues": [],
"source_intake_issues": [],
"word_count": 884
}
],
"pass": false,
"passes": 0,
"scores": [
{
"behavioral_rule_schema_issues": [],
"broad_semantic_issues": [],
"concepts": {
"bounded_intake_tier": false,
"real_source_identity": false,
"staged_review_apply": true,
"staging_without_apply_authorization": false
},
"current_schema_overclaims": [
"sources_unshipped_fields"
],
"custom_signals": {},
"dimension": "autonomous_source_intake_boundary",
"invalid_count_invariant_detected": false,
"legacy_pass": false,
"legacy_signals": {
"artifact": true,
"authorization": true,
"no_overclaim": false,
"staging_or_review": false
},
"max_response_words": 220,
"overclaim_detected": false,
"pass": false,
"prompt_id": "OOS-14",
"proposal_readiness_issues": [],
"response_issue_detected": false,
"response_too_long": true,
"source_evidence_semantic_issues": [],
"source_intake_issues": [],
"word_count": 372
},
{
"behavioral_rule_schema_issues": [],
"broad_semantic_issues": [],
"concepts": {
"behavioral_rule_storage": false,
"heterogeneous_types": true,
"receipt": false,
"reviewed_policy_apply": false,
"staged_review_apply": true
},
"current_schema_overclaims": [],
"custom_signals": {},
"dimension": "database_composition_judgment",
"invalid_count_invariant_detected": false,
"legacy_pass": false,
"legacy_signals": {
"canonical_db": true,
"caveat_retention": false,
"next_action": true,
"no_overclaim": false,
"staging_or_review": false
},
"max_response_words": 220,
"overclaim_detected": false,
"pass": false,
"prompt_id": "OOS-06",
"proposal_readiness_issues": [],
"response_issue_detected": false,
"response_too_long": true,
"source_evidence_semantic_issues": [],
"source_intake_issues": [],
"word_count": 884
}
]
},
"selected_prompt_ids": [
"OOS-14",
"OOS-06"
],
"service_unchanged": true,
"source_results_json": "/Users/user/Documents/Codex/2026-07-09/019f34eb-d297-72d0-b7e2-b222d5515ab9-load/work/teleo-infra-main/docs/reports/leo-working-state-20260709/telegram-handler-m3taversal-oos-targeted-post-deterministic-context-contract-20260713.json",
"temp_profile_removed": true
}

98
docs/reports/leo-working-state-20260709/telegram-handler-m3taversal-oos-targeted-post-deterministic-context-contract-20260713.json Normal file
View file

File diff suppressed because one or more lines are too long

18
docs/reports/leo-working-state-20260709/telegram-handler-m3taversal-oos-targeted-post-deterministic-context-contract-20260713.md Normal file
View file

@ -0,0 +1,18 @@
# Working Leo Targeted Deterministic Context Contract
Generated UTC: `2026-07-13T10:28:02.926071+00:00`
Safe pass: `False`
Prompts: `0/2`
DB counts changed: `False`
Service unchanged: `True`
Temporary profile removed: `True`
Posted to Telegram: `False`
## Prompt Scores
- `OOS-14` / `autonomous_source_intake_boundary`: `pass=False`, words=`372`
- `OOS-06` / `database_composition_judgment`: `pass=False`, words=`884`
## Claim Ceiling
This proves the unchanged broad questions through the live VPS GatewayRunner using the deployed database-driven context contract. It does not prove human-visible Telegram delivery or production DB apply.

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

@ -0,0 +1,140 @@
"""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)

6
hermes-agent/leoclean-plugins/vps/leo-db-context/plugin.yaml Normal file
View file

@ -0,0 +1,6 @@
name: leo-db-context
version: 1.0.0
description: Inject query-specific read-only Teleo database contracts before each Leo model turn.
author: living-ip
provides_hooks:
- pre_llm_call

26
scripts/run_leo_direct_claim_handler_suite.py
View file

@ -179,10 +179,25 @@ def write_report_snapshot(report):
pass
def read_database_context_trace(path):
if not path or not path.exists():
return []
records = []
for line in path.read_text(encoding="utf-8").splitlines():
try:
value = json.loads(line)
except json.JSONDecodeError:
continue
if isinstance(value, dict):
records.append(value)
return records
async def run_suite():
before_service = service_state()
before_counts = db_counts()
temp_profile = None
context_trace_path = None
temp_removed = False
report = {
"generated_at_utc": datetime.now(timezone.utc).isoformat(),
@ -212,6 +227,8 @@ async def run_suite():
temp_profile = copy_profile()
os.environ["HERMES_HOME"] = str(temp_profile)
os.environ["HOME"] = "/home/teleo"
context_trace_path = temp_profile / "state" / "leo-db-context-trace.jsonl"
os.environ["LEO_DB_CONTEXT_TRACE_PATH"] = str(context_trace_path)
sys.path.insert(0, str(AGENT_ROOT))
from gateway.config import Platform
@ -245,6 +262,7 @@ async def run_suite():
results = []
for index, prompt in enumerate(PROMPTS, start=1):
trace_start = len(read_database_context_trace(context_trace_path))
event = MessageEvent(
text=prompt["message"],
message_type=MessageType.TEXT,
@ -267,6 +285,7 @@ async def run_suite():
"mutates_kb": False,
"evidence_tier": "live_vps_gatewayrunner_temp_profile",
"claim_ceiling": "Live VPS GatewayRunner reply from temp leoclean profile; no Telegram post; no production DB apply authorized.",
"database_context_trace": read_database_context_trace(context_trace_path)[trace_start:],
}
)
report["results"] = results
@ -290,10 +309,17 @@ async def run_suite():
"after": after_service,
"unchanged_from_preexisting_live_readback": before_service == after_service,
}
database_context_trace = read_database_context_trace(context_trace_path)
if temp_profile is not None:
tmp_root = temp_profile.parent
temp_removed = remove_tree(tmp_root)
report["temp_profile_removed"] = temp_removed
report["database_context_trace"] = database_context_trace
report["database_context_injection_count"] = len(report["database_context_trace"])
report["database_context_all_ok"] = bool(report["database_context_trace"]) and all(
item.get("status") == "ok" and item.get("injected") is True
for item in report["database_context_trace"]
)
write_report_snapshot(report)

6
scripts/run_leo_m3taversal_oos_handler_suite.py
View file

@ -31,6 +31,8 @@ def write_score_markdown(path: Path, report: dict[str, Any]) -> None:
f"DB counts changed: `{report['db_counts_changed']}`",
f"Service unchanged: `{report['service_unchanged']}`",
f"Temporary profile removed: `{report['temp_profile_removed']}`",
f"Database context injections: `{report['database_context_injection_count']}`",
f"Database context all OK: `{report['database_context_all_ok']}`",
f"Posted to Telegram: `{report['posted_to_telegram']}`",
"",
"## Prompt Scores",
@ -64,6 +66,8 @@ def build_score_report(report: dict[str, Any], *, memory_token: str) -> dict[str
"db_counts_changed": report.get("db_counts_changed"),
"service_unchanged": service_unchanged,
"temp_profile_removed": report.get("temp_profile_removed"),
"database_context_injection_count": report.get("database_context_injection_count"),
"database_context_all_ok": report.get("database_context_all_ok"),
"posted_to_telegram": report.get("posted_to_telegram"),
"production_db_apply_ran": False,
}
@ -76,6 +80,8 @@ def score_report_passes(report: dict[str, Any], score_report: dict[str, Any]) ->
and report.get("db_counts_changed") is False
and score_report["service_unchanged"]
and report.get("temp_profile_removed") is True
and report.get("database_context_all_ok") is True
and int(report.get("database_context_injection_count") or 0) >= len(report.get("results") or [])
and report.get("posted_to_telegram") is False
and score_report["score"]["pass"]
)

95
tests/test_hermes_leoclean_db_context_plugin.py Normal file
View file

@ -0,0 +1,95 @@
"""Tests for automatic live database-contract injection into Leo turns."""
from __future__ import annotations
import importlib.util
import json
import subprocess
from pathlib import Path
from types import SimpleNamespace
ROOT = Path(__file__).resolve().parents[1]
PLUGIN = ROOT / "hermes-agent" / "leoclean-plugins" / "vps" / "leo-db-context" / "__init__.py"
def load_plugin():
spec = importlib.util.spec_from_file_location("leo_db_context_plugin", PLUGIN)
assert spec and spec.loader
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def test_plugin_registers_pre_llm_call_hook() -> None:
module = load_plugin()
registered = []
module.register(SimpleNamespace(register_hook=lambda name, callback: registered.append((name, callback))))
assert len(registered) == 1
assert registered[0][0] == "pre_llm_call"
def test_plugin_injects_only_operational_contracts_and_writes_redacted_trace(
tmp_path: Path, monkeypatch
) -> None:
module = load_plugin()
home = tmp_path / "profile"
kb_tool = home / "bin" / "kb_tool.py"
kb_tool.parent.mkdir(parents=True)
kb_tool.write_text("# fixture\n", encoding="utf-8")
trace = tmp_path / "trace.jsonl"
monkeypatch.setenv("LEO_DB_CONTEXT_TRACE_PATH", str(trace))
payload = {
"query": "must not be injected",
"claims": [{"text": "must not be injected"}],
"operational_contracts": [
{"id": "reply_budget", "hard_max_words": 220},
{"id": "source_intake", "capability_tier": "build-local compiler only"},
],
}
calls = []
def fake_runner(command, **kwargs):
calls.append((command, kwargs))
return subprocess.CompletedProcess(command, 0, stdout=json.dumps(payload), stderr="")
context = module.build_database_context(
"operator secret-shaped but nonsecret message", hermes_home=home, runner=fake_runner
)
assert "reply_budget" in context
assert "source_intake" in context
assert "build-local compiler only" in context
assert "must not be injected" not in context
assert "operator secret-shaped" not in trace.read_text(encoding="utf-8")
record = json.loads(trace.read_text(encoding="utf-8"))
assert record["status"] == "ok"
assert record["injected"] is True
assert record["contract_ids"] == ["reply_budget", "source_intake"]
command, kwargs = calls[0]
assert command[2:5] == ["--local", "context", "operator secret-shaped but nonsecret message"]
assert command[-6:] == ["--limit", "0", "--context-limit", "0", "--format", "json"]
assert kwargs["timeout"] == module.DEFAULT_TIMEOUT_SECONDS
def test_plugin_fails_closed_when_database_contract_lookup_fails(tmp_path: Path) -> None:
module = load_plugin()
home = tmp_path / "profile"
kb_tool = home / "bin" / "kb_tool.py"
kb_tool.parent.mkdir(parents=True)
kb_tool.write_text("# fixture\n", encoding="utf-8")
def fake_runner(command, **_kwargs):
return subprocess.CompletedProcess(command, 9, stdout="", stderr="database unavailable")
context = module.build_database_context("What is live?", hermes_home=home, runner=fake_runner)
assert 'status="unavailable"' in context
assert "Do not infer current schema" in context
assert "database unavailable" not in context
def test_handler_harness_retains_database_context_proof_fields() -> None:
source = (ROOT / "scripts" / "run_leo_direct_claim_handler_suite.py").read_text(encoding="utf-8")
assert "LEO_DB_CONTEXT_TRACE_PATH" in source
assert '"database_context_trace"' in source
assert '"database_context_injection_count"' in source
assert '"database_context_all_ok"' in source

7
tests/test_teleo_agent_systemd.py
View file

@ -100,8 +100,10 @@ def test_deploy_scripts_sync_leoclean_skills_and_restart_gateway_for_runtime_cha
assert 'LEOCLEAN_BIN_DIR="/home/teleo/.hermes/profiles/leoclean/bin"' in auto_deploy
assert 'LEOCLEAN_SKILLS_DIR="/home/teleo/.hermes/profiles/leoclean/skills"' in auto_deploy
assert 'LEOCLEAN_PLUGINS_DIR="/home/teleo/.hermes/profiles/leoclean/plugins"' in auto_deploy
assert 'hermes-agent/leoclean-bin/' in auto_deploy
assert 'hermes-agent/leoclean-skills/vps/' in auto_deploy
assert 'hermes-agent/leoclean-plugins/vps/' in auto_deploy
assert 'deploy/leoclean-gateway-restart-required.sh' in auto_deploy
assert 'add_restart_if_unit_active leoclean-gateway' in auto_deploy
assert 'sudo systemctl restart "$svc"' in auto_deploy
@ -110,8 +112,10 @@ def test_deploy_scripts_sync_leoclean_skills_and_restart_gateway_for_runtime_cha
assert "NOPASSWD: /bin/systemctl restart leoclean-gateway" in sudoers
assert 'VPS_LEOCLEAN_BIN="/home/teleo/.hermes/profiles/leoclean/bin"' in manual_deploy
assert 'VPS_LEOCLEAN_SKILLS="/home/teleo/.hermes/profiles/leoclean/skills"' in manual_deploy
assert 'VPS_LEOCLEAN_PLUGINS="/home/teleo/.hermes/profiles/leoclean/plugins"' in manual_deploy
assert 'hermes-agent/leoclean-bin/' in manual_deploy
assert 'hermes-agent/leoclean-skills/vps/' in manual_deploy
assert 'hermes-agent/leoclean-plugins/vps/' in manual_deploy
assert 'systemctl is-active --quiet leoclean-gateway.service' in manual_deploy
@ -129,6 +133,9 @@ def test_auto_deploy_restarts_gateway_for_leoclean_runtime_changes():
assert _leoclean_gateway_restart_required(
"hermes-agent/leoclean-skills/vps/teleo-kb-bridge/runtime_helper.py"
)
assert _leoclean_gateway_restart_required(
"hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py"
)
def test_auto_deploy_sudoers_installer_is_root_only_and_visudo_checked():