diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8a725e6..2b232cd 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -70,6 +70,7 @@ jobs: scripts/build_working_leo_m3taversal_outcome_sandbox.py \ scripts/leo_behavior_manifest.py \ scripts/leo_tool_trace.py \ + scripts/compile_leoclean_nosend_runtime.py \ scripts/replay_decision_engine_eval.py \ scripts/prove_phase1b_local.py \ scripts/run_gcp_generated_db_direct_claim_suite.py \ @@ -81,6 +82,7 @@ jobs: scripts/run_leo_clone_bound_handler_checkpoint.py \ scripts/working_leo_m3taversal_oos_benchmark.py \ scripts/working_leo_open_ended_benchmark.py \ + scripts/verify_leoclean_nosend_runtime.py \ tests/test_agent_routing.py \ tests/test_attest_gcp_reasoning_compute.py \ tests/test_assemble_telegram_visible_direct_claim_capture_receipt.py \ @@ -104,7 +106,9 @@ jobs: tests/test_gcp_generated_db_working_leo_suite.py \ tests/test_gcp_generated_db_blind_claim_canary.py \ tests/test_hermes_leoclean_kb_bridge_source.py \ + tests/test_hermes_leoclean_db_context_plugin.py \ tests/test_hermes_leoclean_skill_surfaces.py \ + tests/test_leoclean_nosend_runtime.py \ tests/test_leo_behavior_manifest.py \ tests/test_leo_tool_trace.py \ tests/test_verify_leo_db_first_oos_canary.py \ @@ -134,7 +138,8 @@ jobs: ops/backup_vps_sqlite_kb.sh \ ops/provision_gcp_leoclean_runtime_role.sh \ ops/run_gcp_cloudsql_restore_drill.sh \ - ops/run_sqlite_postgres_restore_canary.sh + ops/run_sqlite_postgres_restore_canary.sh \ + scripts/run_leoclean_nosend_runtime_canary.sh test: name: Unit tests @@ -205,6 +210,27 @@ jobs: .crabbox-results/decision-engine-eval.json if-no-files-found: error + leoclean-nosend-runtime: + name: Leoclean no-send runtime + runs-on: ubuntu-latest + timeout-minutes: 20 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.11.9" + - name: Install exact uv toolchain + run: python -m pip install 'uv==0.9.30' + - name: Compile and exercise release artifact + run: scripts/run_leoclean_nosend_runtime_canary.sh + - name: Upload no-send runtime receipt + if: always() + uses: actions/upload-artifact@v4 + with: + name: teleo-infrastructure-leoclean-nosend-runtime + path: .crabbox-results/leoclean-nosend-runtime.json + if-no-files-found: error + phase1b-local-proof: name: Phase 1B local proof runs-on: ubuntu-latest diff --git a/docs/gcp-leoclean-nosend-runtime.md b/docs/gcp-leoclean-nosend-runtime.md new file mode 100644 index 0000000..724762d --- /dev/null +++ b/docs/gcp-leoclean-nosend-runtime.md @@ -0,0 +1,96 @@ +# GCP leoclean no-send runtime + +## Outcome + +This slice builds and verifies a reproducible Hermes/leoclean artifact for an +isolated GCP staging service that has no messaging authority. It retains the +reviewed knowledge-reading and proposal-staging surface while making Telegram +or any other outbound transport structurally unavailable. + +The artifact is pinned to Hermes commit +`b2f477a30b3c05d0f383c543af98496ae8a96070`, its exact upstream archive and +dependency lock, the two LivingIP compatibility patches, the leoclean profile +inputs, and the Teleo Git revision that compiled it. + +## Enforced runtime boundary + +- The effective Hermes registry contains exactly `skills_list`, `skill_view`, + and a restricted `terminal` tool. +- `send_message`, `process`, delegation, cron management, memory mutation, and + skill mutation are absent and direct dispatch is rejected. +- Persistent memory, user-profile memory, automatic memory review, and + automatic skill review are independently disabled in the exact profile; + positive or missing review intervals fail verification. +- Cheap-model routing is disabled rather than claimed through an inert config + shape; the release probe verifies the pinned Hermes effective configuration. +- GCP-only Cloud SQL and context adapters live under the runtime tree. The + shared VPS bridge and plugin remain unchanged, so merging this slice neither + syncs nor restarts the VPS service. +- Automatic context currently supports review-only candidate composition and + exact claim/evidence challenges with no extra database queries. Broader + database-lifecycle questions fail closed until the parity slice adds a + separately analyzed Cloud SQL query surface. +- The terminal handler can execute only the artifact's `teleo-kb` wrapper and + its reviewed read/proposal commands. It invokes an argument vector without a + shell, rejects database-identity overrides, bounds time and output, and gives + the child a minimal environment. +- The gateway starts with zero platform adapters and zero connected platforms. + Its tool registry is checked again after startup and model-tool discovery, + then the gateway is stopped. +- The only user plugin is `leo-db-context`; it contributes the reviewed pre- and + post-model hooks and no tools. The only gateway event hook is Hermes's built-in + startup hook, while `BOOT.md` is forbidden from the profile. +- Transport credentials, project plugins, environment injection, extra startup + controls, symlinks, and unbound artifact files fail closed. +- The artifact binds its metadata identity check to the non-production + `sa-teleo-staging-vm` service account. The shared helper retains its existing + production default only when this root-controlled artifact manifest is absent. +- Leo can stage a proposal only through + `kb_stage.stage_leoclean_proposal(...)`; this slice does not grant direct + canonical or staging-table writes. + +## Reproduce the release proof + +Run from a clean Git checkout with network access for the exact source archive +and locked Python wheels: + +```bash +python3 -m pip install 'uv==0.9.30' +scripts/run_leoclean_nosend_runtime_canary.sh +``` + +The canary: + +1. compiles the commit-bound artifact twice, rejects dirty relevant sources, + and requires identical manifests and content hashes; +2. creates an exact CPython 3.11.9 environment; +3. installs only dependencies resolved by the pinned Hermes `uv.lock`; +4. checks that the lock is current; +5. starts and stops the effective no-send gateway with a synthetic provider + credential, proving the credential value is absent from captured output and + the verification receipt; and +6. writes `.crabbox-results/leoclean-nosend-runtime.json` only after release + verification passes. + +GitHub Actions runs this same wrapper and retains the receipt as a CI artifact. + +## Claim ceiling and next slices + +This proof establishes reproducible source and a structurally no-send runtime +surface. It does **not** prove GCP IAM, Secret Manager, Cloud SQL identity, +deployment or restart behavior, model behavior, database parity, Telegram +delivery, or production readiness. + +Those remain separate rollback boundaries: + +1. bind a dedicated non-production service account to the scoped Cloud SQL + secret and `leoclean_kb_runtime` database role; +2. package the artifact with its operating-system dependencies and deploy the + isolated GCP service; +3. prove the running service's effective identity, allowed reads and proposal + staging, denied writes/escalation, receipts, and restart behavior; +4. add an internal no-send prompt ingress and run behavioral/database parity; +5. connect the protected Observatory to canonical claims, proposals, pgvector + projection, and runtime receipts; and +6. cut over only after soak and restore testing, then revoke the VPS credentials + and retire the VPS. diff --git a/hermes-agent/runtime/leoclean-nosend/bootstrap.py b/hermes-agent/runtime/leoclean-nosend/bootstrap.py new file mode 100644 index 0000000..a6cad4f --- /dev/null +++ b/hermes-agent/runtime/leoclean-nosend/bootstrap.py @@ -0,0 +1,794 @@ +#!/usr/bin/env python3 +"""Start or inspect the structurally no-send Hermes/leoclean staging runtime.""" + +from __future__ import annotations + +import argparse +import asyncio +import hashlib +import importlib +import json +import os +import platform +import shlex +import signal +import stat +import subprocess +import sys +from pathlib import Path +from typing import Any + + +class NoSendContractError(RuntimeError): + """The runtime does not satisfy its no-send contract.""" + + +def _require(condition: bool, message: str) -> None: + if not condition: + raise NoSendContractError(message) + + +class _SealedToolMap(dict[str, Any]): + """Read-only registry map after the reviewed tools are installed.""" + + @staticmethod + def _deny(*_args: Any, **_kwargs: Any) -> None: + raise NoSendContractError("Hermes tool registry is sealed by the no-send runtime") + + __setitem__ = _deny + __delitem__ = _deny + clear = _deny + pop = _deny + popitem = _deny + setdefault = _deny + update = _deny + + +def _load_json(path: Path) -> dict[str, Any]: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise NoSendContractError(f"cannot load {path.name}: {type(exc).__name__}") from exc + _require(isinstance(value, dict), f"{path.name} must be a JSON object") + return value + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _runtime_root() -> Path: + return Path(__file__).resolve().parent + + +def _validate_contract(value: dict[str, Any]) -> None: + _require( + set(value) + == { + "schema", + "runtime_mode", + "transport_authority", + "service_mode", + "handler_platform", + "plugins", + "toolset", + "profile", + "environment", + "database", + "claim_ceiling", + }, + "no-send runtime contract fields are not exact", + ) + _require( + value.get("schema") == "livingip.leocleanNoSendRuntimeContract.v1", + "no-send runtime contract schema is invalid", + ) + _require(value.get("runtime_mode") == "gcp_staging_no_send", "runtime mode is not no-send staging") + _require(value.get("transport_authority") == "absent", "transport authority is not absent") + _require(value.get("service_mode") == "gateway_without_platform_adapters", "service mode is not no-send") + _require(value.get("handler_platform") == "api_server", "handler platform is not the reviewed surface") + _require( + value.get("plugins") + == { + "allowed": [ + { + "name": "leo-db-context", + "source": "user", + "tools": [], + "hooks": ["post_llm_call", "pre_llm_call"], + } + ] + }, + "plugin contract is not exact", + ) + toolset = value.get("toolset") or {} + _require( + toolset.get("allowed_tools") == ["skills_list", "skill_view", "terminal"], + "allowed tool contract is not exact", + ) + _require( + set(toolset.get("forbidden_tools") or []) + == {"cronjob", "delegate_task", "memory", "process", "send_message", "skill_manage"}, + "forbidden tool contract is not exact", + ) + _require( + set(toolset.get("terminal_commands") or []) + == { + "context", + "decision-matrix-status", + "edges", + "evidence", + "list-proposals", + "propose-core-change", + "propose-edge", + "search", + "search-proposals", + "show", + "show-proposal", + "status", + }, + "terminal command contract is not exact", + ) + profile = value.get("profile") or {} + _require( + set(profile.get("template_entries") or []) == {"bin", "config.yaml", "plugins", "skills"}, + "profile template contract is not exact", + ) + _require( + set(profile.get("required_runtime_entries") or []) + == {"SOUL.md", "identity-lock.json", "identity-manifest.json"}, + "runtime identity contract is not exact", + ) + _require( + set(profile.get("forbidden_entries") or []) + == {".env", "BOOT.md", "auth.json", "gateway.json", "hooks", "processes.json", "secrets"}, + "forbidden profile contract is not exact", + ) + learning_policy = profile.get("learning_policy") or {} + _require( + set(learning_policy) + == { + "automatic_memory_review_interval", + "automatic_skill_review_interval", + "memory_enabled", + "user_profile_enabled", + } + and learning_policy.get("memory_enabled") is False + and learning_policy.get("user_profile_enabled") is False + and type(learning_policy.get("automatic_memory_review_interval")) is int + and learning_policy.get("automatic_memory_review_interval") == 0 + and type(learning_policy.get("automatic_skill_review_interval")) is int + and learning_policy.get("automatic_skill_review_interval") == 0, + "learning policy contract is not exact", + ) + routing_policy = profile.get("routing_policy") or {} + _require( + set(routing_policy) == {"cheap_model", "smart_model_routing"} + and routing_policy.get("cheap_model") == "absent" + and routing_policy.get("smart_model_routing") is False, + "routing policy contract is not exact", + ) + environment = value.get("environment") or {} + _require( + environment.get("allowed_provider_credentials") == ["OPENROUTER_API_KEY"], + "provider credential contract is not exact", + ) + _require( + set(environment.get("allowed_runtime_variables") or []) + == {"HERMES_HOME", "PYTHONDONTWRITEBYTECODE", "PYTHONNOUSERSITE", "PYTHONSAFEPATH"}, + "runtime environment allowlist is not exact", + ) + required_forbidden_exact = { + "ALL_PROXY", + "BASH_ENV", + "BOT_TOKEN", + "ENV", + "GATEWAY_ALLOW_ALL_USERS", + "GATEWAY_ALLOWED_USERS", + "GOOGLE_APPLICATION_CREDENTIALS", + "HERMES_ENABLE_PROJECT_PLUGINS", + "HERMES_SESSION_PLATFORM", + "HTTP_PROXY", + "HTTPS_PROXY", + "LD_PRELOAD", + "PYTHONHOME", + "PYTHONPATH", + "TELEGRAM_BOT_TOKEN", + } + _require( + set(environment.get("forbidden_exact") or []) == required_forbidden_exact, + "forbidden environment contract is not exact", + ) + _require( + set(environment.get("forbidden_prefixes") or []) + == { + "CLOUDSDK_", + "DINGTALK_", + "DISCORD_", + "EMAIL_", + "FEISHU_", + "HERMES_", + "LD_", + "MATRIX_", + "MATTERMOST_", + "SIGNAL_", + "SLACK_", + "SMS_", + "TELEGRAM_", + "TWILIO_", + "WEBCOM_", + "WEBHOOK_", + "WHATSAPP_", + }, + "transport environment prefix contract is not exact", + ) + _require( + set(environment.get("forbidden_suffixes") or []) == {"_PASSWORD", "_SECRET", "_TOKEN", "_API_KEY"}, + "secret-shaped environment contract is not exact", + ) + database = value.get("database") or {} + _require( + database + == { + "automatic_context_modes": ["claim_evidence_challenge", "review_only_candidate"], + "canonical_database": "teleo_canonical", + "runtime_role": "leoclean_kb_runtime", + "proposal_function": "kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)", + "direct_canonical_writes": "denied", + "direct_stage_table_writes": "denied", + "proposal_staging": "function_only", + "unsupported_context_policy": "fail_closed", + }, + "database authority contract is not exact", + ) + _require(isinstance(value.get("claim_ceiling"), str) and bool(value["claim_ceiling"]), "claim ceiling is missing") + + +def _contract() -> dict[str, Any]: + value = _load_json(_runtime_root() / "runtime-contract.json") + _require( + value.get("schema") == "livingip.leocleanNoSendRuntimeContract.v1", + "unsupported no-send runtime contract", + ) + _validate_contract(value) + return value + + +def _validate_environment(contract: dict[str, Any]) -> None: + policy = contract["environment"] + allowed_provider = set(policy["allowed_provider_credentials"]) + allowed_runtime = set(policy["allowed_runtime_variables"]) + forbidden_exact = set(policy["forbidden_exact"]) + forbidden_prefixes = tuple(policy["forbidden_prefixes"]) + forbidden_suffixes = tuple(policy["forbidden_suffixes"]) + violations: list[str] = [] + for name, value in os.environ.items(): + if not value: + continue + if name in allowed_provider or name in allowed_runtime: + continue + if name in forbidden_exact or name.startswith(forbidden_prefixes) or name.endswith(forbidden_suffixes): + violations.append(name) + _require(not violations, f"forbidden transport environment is present: {sorted(violations)}") + + +def _validate_profile(profile: Path, contract: dict[str, Any], *, template: bool) -> None: + _require(profile.is_dir() and not profile.is_symlink(), "profile root is missing or unsafe") + forbidden = set(contract["profile"]["forbidden_entries"]) + actual = {path.name for path in profile.iterdir()} + present_forbidden = sorted(actual & forbidden) + _require(not present_forbidden, f"forbidden profile entries are present: {present_forbidden}") + _require(not any(path.is_symlink() for path in profile.rglob("*")), "profile contains a symlink") + immutable_entries = set(contract["profile"]["template_entries"]) + expected_template = _runtime_root() / "profile-template" + _require(expected_template.is_dir(), "bound profile template is missing") + for entry in sorted(immutable_entries): + expected = expected_template / entry + observed = profile / entry + _require(expected.exists() and observed.exists(), f"immutable profile entry is missing: {entry}") + _require(expected.is_dir() == observed.is_dir(), f"immutable profile entry type drifted: {entry}") + if expected.is_file(): + _require(_sha256(expected) == _sha256(observed), f"immutable profile file drifted: {entry}") + _require( + stat.S_IMODE(expected.stat().st_mode) == stat.S_IMODE(observed.stat().st_mode), + f"immutable profile file mode drifted: {entry}", + ) + continue + expected_files = { + path.relative_to(expected).as_posix(): ( + _sha256(path), + stat.S_IMODE(path.stat().st_mode), + ) + for path in expected.rglob("*") + if path.is_file() + } + observed_files = { + path.relative_to(observed).as_posix(): ( + _sha256(path), + stat.S_IMODE(path.stat().st_mode), + ) + for path in observed.rglob("*") + if path.is_file() + } + _require(observed_files == expected_files, f"immutable profile tree drifted: {entry}") + if template: + _require(actual == immutable_entries, "profile template entry set is not exact") + else: + required_identity = set(contract["profile"]["required_runtime_entries"]) + state_policy = contract["profile"]["allowed_runtime_state"] + state_directories = set(state_policy["directories"]) + state_files = set(state_policy["files"]) + allowed = immutable_entries | required_identity | state_directories | state_files + _require(required_identity <= actual, "compiled runtime profile is incomplete") + _require(actual <= allowed, f"compiled runtime profile has unapproved entries: {sorted(actual - allowed)}") + for name in sorted(required_identity): + entry = profile / name + _require( + entry.is_file() and entry.stat().st_size <= 1024 * 1024, f"runtime identity entry is invalid: {name}" + ) + for name in sorted(actual & state_directories): + _require((profile / name).is_dir(), f"runtime state entry must be a directory: {name}") + for name in sorted(actual & state_files): + _require((profile / name).is_file(), f"runtime state entry must be a file: {name}") + + +def _load_profile_config(profile: Path, contract: dict[str, Any]) -> dict[str, Any]: + try: + import yaml + + value = yaml.safe_load((profile / "config.yaml").read_text(encoding="utf-8")) or {} + except (OSError, TypeError, ValueError) as exc: + raise NoSendContractError(f"cannot load config.yaml: {type(exc).__name__}") from exc + _require(isinstance(value, dict), "config.yaml must be a mapping") + expected_top_level = { + "agent", + "display", + "known_plugin_toolsets", + "mcp_servers", + "memory", + "model", + "platform_toolsets", + "skills", + "smart_model_routing", + } + _require(set(value) == expected_top_level, "config.yaml top-level fields are not exact") + _require(value.get("mcp_servers") == {}, "MCP servers are forbidden in no-send staging") + memory = value.get("memory") or {} + _require( + set(memory) == {"enabled", "memory_enabled", "nudge_interval", "user_profile_enabled"} + and memory.get("enabled") is False + and memory.get("memory_enabled") is False + and memory.get("user_profile_enabled") is False + and type(memory.get("nudge_interval")) is int + and memory.get("nudge_interval") == 0, + "persistent memory and automatic memory review must be disabled", + ) + skills = value.get("skills") or {} + _require( + set(skills) == {"creation_nudge_interval"} + and type(skills.get("creation_nudge_interval")) is int + and skills.get("creation_nudge_interval") == 0, + "automatic skill review must be disabled", + ) + _require(value.get("smart_model_routing") is False, "smart-model routing must be disabled") + expected_toolset = contract["toolset"]["name"] + _require( + value.get("platform_toolsets") == {"api_server": [expected_toolset]}, + "handler toolset binding is not exact", + ) + _require( + value.get("known_plugin_toolsets") == {"api_server": []}, + "plugin toolset allowlist is not exact", + ) + encoded = json.dumps(value, sort_keys=True).lower() + for marker in ( + "telegram", + "discord", + "whatsapp", + "slack", + "signal", + "mattermost", + "matrix", + "dingtalk", + "feishu", + "wecom", + "webhook", + "bot_token", + "send_message", + ): + _require(marker not in encoded, f"transport marker is forbidden in config.yaml: {marker}") + return value + + +def _load_effective_policy(contract: dict[str, Any]) -> dict[str, Any]: + """Prove what the pinned Hermes loader will give the actual agent runtime.""" + + try: + from agent.smart_model_routing import choose_cheap_model_route + from hermes_cli.config import load_config + + effective = load_config() + except Exception as exc: + raise NoSendContractError(f"cannot load effective Hermes config: {type(exc).__name__}") from exc + _require(isinstance(effective, dict), "effective Hermes config must be a mapping") + learning_policy = contract["profile"]["learning_policy"] + memory = effective.get("memory") or {} + skills = effective.get("skills") or {} + _require(isinstance(memory, dict) and isinstance(skills, dict), "effective learning policy is invalid") + _require(memory.get("memory_enabled") is False, "effective Hermes memory policy is not disabled") + _require(memory.get("user_profile_enabled") is False, "effective Hermes user profile is not disabled") + memory_interval = memory.get("nudge_interval") + _require( + type(memory_interval) is int and memory_interval == learning_policy["automatic_memory_review_interval"], + "effective Hermes memory-review interval is not disabled", + ) + skill_interval = skills.get("creation_nudge_interval") + _require( + type(skill_interval) is int and skill_interval == learning_policy["automatic_skill_review_interval"], + "effective Hermes skill-review policy is not disabled", + ) + _require(effective.get("smart_model_routing") is False, "effective Hermes smart routing is not disabled") + _require( + choose_cheap_model_route("what time is it?", effective.get("smart_model_routing")) is None, + "effective Hermes config selected a cheap-model route", + ) + return { + "memory_enabled": memory.get("memory_enabled"), + "user_profile_enabled": memory.get("user_profile_enabled"), + "memory_nudge_interval": memory.get("nudge_interval"), + "skill_creation_nudge_interval": skills.get("creation_nudge_interval"), + "smart_model_routing": False, + "cheap_model_route": None, + } + + +def _restricted_terminal_handler( + profile: Path, + allowed_commands: set[str], + tool_args: dict[str, Any], + **_kwargs: Any, +) -> str: + if not isinstance(tool_args, dict): + return json.dumps({"output": "", "exit_code": 126, "error": "terminal arguments must be an object"}) + unexpected = sorted(set(tool_args) - {"command", "timeout"}) + if unexpected: + return json.dumps({"output": "", "exit_code": 126, "error": "terminal arguments contain unsupported fields"}) + command = tool_args.get("command") + if not isinstance(command, str) or not command.strip(): + return json.dumps({"output": "", "exit_code": 126, "error": "terminal command is required"}) + if "\x00" in command or "\n" in command or "\r" in command: + return json.dumps( + {"output": "", "exit_code": 126, "error": "terminal command contains a forbidden control character"} + ) + try: + argv = shlex.split(command, posix=True) + except ValueError as exc: + return json.dumps({"output": "", "exit_code": 126, "error": f"invalid terminal command: {exc}"}) + wrapper = (profile / "bin" / "teleo-kb").resolve() + if not argv or argv[0] not in {"teleo-kb", str(wrapper)}: + return json.dumps({"output": "", "exit_code": 126, "error": "only the reviewed teleo-kb wrapper is allowed"}) + if len(argv) < 2 or argv[1] not in allowed_commands: + return json.dumps({"output": "", "exit_code": 126, "error": "teleo-kb command is outside the allowlist"}) + forbidden_tokens = {";", "&&", "||", "|", ">", ">>", "<", "<<", "&"} + forbidden_overrides = { + "--canonical-db", + "--credential-mode", + "--db", + "--host", + "--password-secret", + "--port", + "--project", + "--user", + } + if any(token in forbidden_tokens for token in argv[2:]): + return json.dumps({"output": "", "exit_code": 126, "error": "terminal command contains a shell control token"}) + if any( + token in forbidden_overrides or any(token.startswith(f"{option}=") for option in forbidden_overrides) + for token in argv[2:] + ): + return json.dumps({"output": "", "exit_code": 126, "error": "terminal command contains an identity override"}) + try: + timeout = int(tool_args.get("timeout") or 60) + except (TypeError, ValueError): + return json.dumps({"output": "", "exit_code": 126, "error": "terminal timeout must be an integer"}) + timeout = min(max(timeout, 1), 120) + child_environment = { + "HOME": str(profile / "state"), + "HERMES_HOME": str(profile), + "LANG": "C", + "LC_ALL": "C", + "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", + } + try: + result = subprocess.run( + [str(wrapper), *argv[1:]], + cwd=profile, + env=child_environment, + text=True, + capture_output=True, + check=False, + timeout=timeout, + ) + except subprocess.TimeoutExpired: + return json.dumps({"output": "", "exit_code": 124, "error": "teleo-kb command timed out"}) + return json.dumps( + { + "output": result.stdout[:65536], + "exit_code": result.returncode, + "error": (result.stderr[:65536] or None), + "truncated": len(result.stdout) > 65536 or len(result.stderr) > 65536, + } + ) + + +def _restricted_terminal_schema(commands: list[str]) -> dict[str, Any]: + return { + "name": "terminal", + "description": "Run one reviewed teleo-kb knowledge command without a shell.", + "parameters": { + "type": "object", + "properties": { + "command": { + "type": "string", + "description": f"teleo-kb followed by one of: {', '.join(commands)}", + }, + "timeout": { + "type": "integer", + "minimum": 1, + "maximum": 120, + }, + }, + "required": ["command"], + "additionalProperties": False, + }, + } + + +def _plugin_inventory(contract: dict[str, Any]) -> list[dict[str, Any]]: + from hermes_cli.plugins import get_plugin_manager + + manager = get_plugin_manager() + observed: list[dict[str, Any]] = [] + for name, loaded in sorted(manager._plugins.items()): + observed.append( + { + "name": name, + "source": loaded.manifest.source, + "tools": sorted(loaded.tools_registered), + "hooks": sorted(loaded.hooks_registered), + "enabled": loaded.enabled, + "error": loaded.error, + } + ) + expected = [ + { + **item, + "tools": sorted(item["tools"]), + "hooks": sorted(item["hooks"]), + "enabled": True, + "error": None, + } + for item in contract["plugins"]["allowed"] + ] + _require(observed == expected, "effective Hermes plugin inventory is not exact") + return observed + + +def _assert_tool_registry(profile: Path, contract: dict[str, Any]) -> dict[str, Any]: + from tools.registry import registry + + allowed_tools = sorted(contract["toolset"]["allowed_tools"]) + _require(sorted(registry._tools) == allowed_tools, "effective Hermes tool registry is not exact") + _require(isinstance(registry._tools, _SealedToolMap), "Hermes tool registry is not sealed") + terminal_entry = registry._tools.get("terminal") + _require(terminal_entry is not None, "restricted terminal is not registered") + expected_marker = f"livingip-nosend-terminal:{profile.resolve()}" + _require( + getattr(terminal_entry.handler, "_livingip_no_send_handler", None) == expected_marker, + "restricted terminal handler identity drifted", + ) + _require( + terminal_entry.schema == _restricted_terminal_schema(contract["toolset"]["terminal_commands"]), + "restricted terminal schema drifted", + ) + for forbidden in contract["toolset"]["forbidden_tools"]: + _require(forbidden not in registry._tools, f"forbidden Hermes tool remains registered: {forbidden}") + dispatched = json.loads(registry.dispatch(forbidden, {})) + _require( + dispatched == {"error": f"Unknown tool: {forbidden}"}, f"forbidden tool dispatch is not denied: {forbidden}" + ) + return { + "toolset": contract["toolset"]["name"], + "tools": sorted(registry._tools), + "send_message_present": "send_message" in registry._tools, + "registry_sealed": True, + "terminal_handler": "livingip.restricted_teleo_kb.v1", + "terminal_restricted_to": "profile/bin/teleo-kb", + "plugins": _plugin_inventory(contract), + } + + +def _prepare_tool_surface(profile: Path, config: dict[str, Any], contract: dict[str, Any]) -> dict[str, Any]: + import toolsets + from hermes_cli import tools_config + from tools.registry import registry + + importlib.import_module("tools.skills_tool") + importlib.import_module("tools.terminal_tool") + importlib.import_module("tools.process_registry") + + toolset_name = contract["toolset"]["name"] + allowed_tools = list(contract["toolset"]["allowed_tools"]) + toolsets.TOOLSETS[toolset_name] = { + "description": "LivingIP no-send Cloud SQL knowledge tools", + "tools": allowed_tools, + "includes": [], + } + resolved = tools_config._get_platform_tools(config, contract["handler_platform"]) + _require(resolved == {toolset_name}, "effective Hermes toolset is not exact") + missing = sorted(name for name in allowed_tools if name not in registry._tools) + _require(not missing, f"required Hermes tools are not registered: {missing}") + _plugin_inventory(contract) + retained = {name: registry._tools[name] for name in allowed_tools} + terminal_entry = retained["terminal"] + + def restricted_terminal(args: dict[str, Any], **kwargs: Any) -> str: + return _restricted_terminal_handler( + profile, + set(contract["toolset"]["terminal_commands"]), + args, + **kwargs, + ) + + restricted_terminal._livingip_no_send_handler = f"livingip-nosend-terminal:{profile.resolve()}" # type: ignore[attr-defined] + terminal_entry.handler = restricted_terminal + terminal_entry.schema = _restricted_terminal_schema(contract["toolset"]["terminal_commands"]) + terminal_entry.check_fn = None + terminal_entry.requires_env = [] + terminal_entry.toolset = toolset_name + for name, entry in retained.items(): + entry.toolset = toolset_name + if name != "terminal": + entry.check_fn = None + entry.requires_env = [] + registry._tools = _SealedToolMap(retained) + registry._toolset_checks = {} + return _assert_tool_registry(profile, contract) + + +def _prepare( + profile: Path, + *, + template: bool, +) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any], dict[str, Any]]: + contract = _contract() + _validate_environment(contract) + _validate_profile(profile, contract, template=template) + os.environ["HERMES_HOME"] = str(profile) + hermes_source = _runtime_root() / "hermes-agent" + _require(hermes_source.is_dir(), "bound Hermes source tree is missing") + if str(hermes_source) not in sys.path: + sys.path.insert(0, str(hermes_source)) + config = _load_profile_config(profile, contract) + effective_policy = _load_effective_policy(contract) + tool_surface = _prepare_tool_surface(profile, config, contract) + return contract, config, tool_surface, effective_policy + + +async def _exercise_runner(runner: Any, profile: Path, contract: dict[str, Any]) -> dict[str, Any]: + started = await runner.start() + _require(started, "Hermes no-send gateway did not start") + post_start = _assert_tool_registry(profile, contract) + hooks = [ + {"name": item.get("name"), "path": item.get("path"), "events": sorted(item.get("events") or [])} + for item in runner.hooks.loaded_hooks + ] + _require( + hooks == [{"name": "boot-md", "path": "(builtin)", "events": ["gateway:startup"]}], + "effective gateway hook inventory is not exact", + ) + importlib.import_module("model_tools") + post_discovery = _assert_tool_registry(profile, contract) + await runner.stop() + _require(not runner.adapters, "gateway retained a transport adapter after stop") + return { + "started": True, + "stopped": True, + "post_start_registry": post_start, + "post_discovery_registry": post_discovery, + "gateway_hooks": hooks, + } + + +def probe(profile: Path, *, template: bool) -> dict[str, Any]: + contract, config, tool_surface, effective_policy = _prepare(profile, template=template) + from gateway.config import GatewayConfig + from gateway.run import GatewayRunner + + gateway_config = GatewayConfig.from_dict({"platforms": {}}) + runner = GatewayRunner(gateway_config) + adapters = getattr(runner, "adapters", None) + _require(isinstance(adapters, dict), "GatewayRunner adapters are not inspectable") + _require(not adapters, "GatewayRunner unexpectedly has a transport adapter") + _require(getattr(runner, "_smart_model_routing", None) == {}, "GatewayRunner retained smart-model routing") + _require(not gateway_config.get_connected_platforms(), "a messaging platform is connected") + _assert_tool_registry(profile, contract) + lifecycle = asyncio.run(_exercise_runner(runner, profile, contract)) + return { + "schema": "livingip.leocleanNoSendRuntimeProbe.v1", + "status": "pass", + "runtime_mode": contract["runtime_mode"], + "config_sha256": _sha256(profile / "config.yaml"), + "gateway_runner": f"{runner.__class__.__module__}.{runner.__class__.__name__}", + "gateway_adapter_count": len(adapters), + "connected_platforms": [], + "tool_surface": tool_surface, + "effective_policy": effective_policy, + "lifecycle": lifecycle, + "provider_credential_names_present": sorted( + name for name in contract["environment"]["allowed_provider_credentials"] if os.getenv(name) + ), + "claim_ceiling": contract["claim_ceiling"], + "model": config["model"]["default"], + "python_version": platform.python_version(), + } + + +async def _run_service(profile: Path) -> bool: + contract, _config, _surface, _effective_policy = _prepare(profile, template=False) + from gateway.config import GatewayConfig + from gateway.run import GatewayRunner + + config = GatewayConfig.from_dict({"platforms": {}}) + runner = GatewayRunner(config) + _require(not runner.adapters, "service cannot start with a transport adapter") + _require(getattr(runner, "_smart_model_routing", None) == {}, "service retained smart-model routing") + _assert_tool_registry(profile, contract) + importlib.import_module("model_tools") + _assert_tool_registry(profile, contract) + + loop = asyncio.get_running_loop() + stop_tasks: set[asyncio.Task[Any]] = set() + + def stop() -> None: + task = asyncio.create_task(runner.stop()) + stop_tasks.add(task) + task.add_done_callback(stop_tasks.discard) + + for sig in (signal.SIGINT, signal.SIGTERM): + try: + loop.add_signal_handler(sig, stop) + except NotImplementedError: + pass + started = await runner.start() + _require(started, "Hermes no-send gateway did not start") + _assert_tool_registry(profile, contract) + await runner.wait_for_shutdown() + return True + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + for name in ("probe", "service"): + child = subparsers.add_parser(name) + child.add_argument("--profile", required=True, type=Path) + if name == "probe": + child.add_argument("--template", action="store_true") + args = parser.parse_args() + try: + if args.command == "probe": + print(json.dumps(probe(args.profile.resolve(), template=args.template), sort_keys=True)) + return 0 + return 0 if asyncio.run(_run_service(args.profile.resolve())) else 1 + except NoSendContractError as exc: + print(json.dumps({"status": "fail", "error": str(exc)}, sort_keys=True), file=sys.stderr) + return 65 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/hermes-agent/runtime/leoclean-nosend/profile-sources/cloudsql_memory_runtime.py b/hermes-agent/runtime/leoclean-nosend/profile-sources/cloudsql_memory_runtime.py new file mode 100644 index 0000000..b8794b0 --- /dev/null +++ b/hermes-agent/runtime/leoclean-nosend/profile-sources/cloudsql_memory_runtime.py @@ -0,0 +1,215 @@ +#!/usr/bin/env python3 +"""GCP-staging entry point over the unchanged shared Cloud SQL bridge.""" + +from __future__ import annotations + +import hashlib +import importlib.util +import json +import stat +import sys +from pathlib import Path +from types import ModuleType +from typing import Any + +RUNTIME_IDENTITY_SCHEMA = "livingip.leocleanGcpRuntimeIdentity.v1" +STAGING_SERVICE_ACCOUNT = "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com" +HERE = Path(__file__).resolve().parent + + +def _load_sibling(name: str, filename: str) -> ModuleType: + path = HERE / filename + if path.is_symlink() or not path.is_file(): + raise SystemExit(f"Bound runtime module was unavailable: {filename}.") + spec = importlib.util.spec_from_file_location(name, path) + if spec is None or spec.loader is None: + raise SystemExit(f"Bound runtime module was unavailable: {filename}.") + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +cloudsql = _load_sibling("livingip_cloudsql_memory_base", "cloudsql_memory_base.py") +contracts = _load_sibling("livingip_operational_contracts", "operational_contracts.py") +_BASE_QUERY_ROWS = cloudsql.query_rows +_BASE_PARSE_ARGS = cloudsql.parse_args + + +def _parse_runtime_identity(body: bytes) -> str: + def reject_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + value: dict[str, Any] = {} + for key, item in pairs: + if key in value: + raise ValueError("duplicate key") + value[key] = item + return value + + try: + value = json.loads(body, object_pairs_hook=reject_duplicates) + except (UnicodeDecodeError, ValueError): + raise SystemExit("Runtime identity manifest was invalid.") from None + if value != { + "expected_service_account": STAGING_SERVICE_ACCOUNT, + "schema": RUNTIME_IDENTITY_SCHEMA, + }: + raise SystemExit("Runtime identity manifest was invalid.") + return STAGING_SERVICE_ACCOUNT + + +def expected_runtime_service_account(path: Path | None = None) -> str: + """Read the exact root-controlled staging identity bound beside this tool.""" + + identity_path = path or (HERE / "gcp-runtime-identity.json") + try: + if identity_path.is_symlink(): + raise OSError + resolved = identity_path.resolve(strict=True) + if resolved != identity_path or not resolved.is_file(): + raise OSError + metadata = resolved.stat() + if metadata.st_uid != 0 or stat.S_IMODE(metadata.st_mode) != 0o644: + raise OSError + body = resolved.read_bytes() + except OSError: + raise SystemExit("Runtime identity manifest was invalid.") from None + if not body or len(body) > 1024: + raise SystemExit("Runtime identity manifest was invalid.") + return _parse_runtime_identity(body) + + +def _parse_args() -> Any: + """Consume bounded contextual-retrieval flags before the shared parser.""" + + filtered = [sys.argv[0]] + retrieval_query: str | None = None + retrieval_anchors: list[str] = [] + arguments = iter(sys.argv[1:]) + for argument in arguments: + if argument == "--retrieval-query": + if retrieval_query is not None: + raise SystemExit("Only one retrieval query is allowed.") + retrieval_query = next(arguments, None) + if retrieval_query is None: + raise SystemExit("Retrieval query value is required.") + continue + if argument.startswith("--retrieval-query="): + if retrieval_query is not None: + raise SystemExit("Only one retrieval query is allowed.") + retrieval_query = argument.split("=", 1)[1] + continue + if argument == "--retrieval-anchor": + anchor = next(arguments, None) + if anchor is None: + raise SystemExit("Retrieval anchor value is required.") + retrieval_anchors.append(anchor) + continue + if argument.startswith("--retrieval-anchor="): + retrieval_anchors.append(argument.split("=", 1)[1]) + continue + filtered.append(argument) + if retrieval_query is not None and (not retrieval_query.strip() or len(retrieval_query) > 16_000): + raise SystemExit("Retrieval query was invalid.") + if len(retrieval_anchors) > 4 or any(not item.strip() or len(item) > 120 for item in retrieval_anchors): + raise SystemExit("Retrieval anchors were invalid.") + original = sys.argv + try: + sys.argv = filtered + parsed = _BASE_PARSE_ARGS() + finally: + sys.argv = original + parsed.retrieval_query = retrieval_query + parsed.retrieval_anchor = retrieval_anchors + return parsed + + +def _claim_mentions_anchor(claim: dict[str, Any], anchors: list[str]) -> bool: + haystack = " ".join([str(claim.get("text") or ""), *(str(item) for item in claim.get("tags") or [])]).casefold() + return any(anchor.casefold() in haystack for anchor in anchors) + + +def _merge_anchor_backfill( + primary: dict[str, Any], + anchor_result: dict[str, Any], + *, + anchors: list[str], + limit: int, +) -> dict[str, Any]: + primary_claims = list(primary.get("claims") or []) + if any(_claim_mentions_anchor(claim, anchors) for claim in primary_claims): + return primary + primary_ids = {str(claim.get("id") or "") for claim in primary_claims} + backfill = [ + claim for claim in list(anchor_result.get("claims") or []) if str(claim.get("id") or "") not in primary_ids + ][: max(1, min(2, limit))] + if not backfill: + return primary + claims = (backfill + primary_claims)[:limit] + context_rows = list(primary.get("context_rows") or []) + include_private = str(primary.get("privacy") or "").startswith("private agent context mode") + claim_hits = [] + for claim in claims: + hit = { + "source_table": "public.claims", + "row_id": claim.get("id"), + "score": claim.get("score"), + "claim_type": claim.get("type"), + "claim_status": claim.get("status"), + "confidence": claim.get("confidence"), + "tags": claim.get("tags") or [], + "evidence_count": claim.get("evidence_count"), + "edge_count": claim.get("edge_count"), + } + if include_private: + hit["excerpt"] = claim.get("text") + claim_hits.append(hit) + context_hits = [hit for hit in list(primary.get("hits") or []) if hit.get("source_table") != "public.claims"] + merged = dict(primary) + merged["claims"] = claims + merged["hits"] = claim_hits + context_hits + merged["hit_count_total"] = len(claims) + len(context_rows) + return merged + + +def _enrich_context(args: Any, query: str, result: dict[str, Any]) -> dict[str, Any]: + """Attach current-main response contracts to the real Cloud SQL rows.""" + + operational_contracts = contracts.operational_contracts(query) + claims = result.get("claims") + if not isinstance(claims, list) or not all(isinstance(item, dict) for item in claims): + raise SystemExit("Cloud SQL context returned an invalid claims payload.") + evidence = {str(claim.get("id") or ""): list(claim.get("evidence") or []) for claim in claims if claim.get("id")} + contracts.bind_claim_evidence_challenge(operational_contracts, claims, evidence) + enriched = dict(result) + enriched["operational_contracts"] = operational_contracts + enriched["compiled_response"] = contracts.compile_operational_response(operational_contracts) + enriched["runtime_contract_mode"] = contracts.classify_contract_mode(query) + return enriched + + +def _query_rows(args: Any, query: str, limit: int) -> dict[str, Any]: + retrieval_query = str(getattr(args, "retrieval_query", None) or query) + retrieval_anchors = list(getattr(args, "retrieval_anchor", None) or []) + result = _BASE_QUERY_ROWS(args, retrieval_query, limit) + if getattr(args, "command", None) == "context" and retrieval_anchors: + anchor_result = _BASE_QUERY_ROWS(args, " ".join(retrieval_anchors), max(1, min(2, limit))) + result = _merge_anchor_backfill(result, anchor_result, anchors=retrieval_anchors, limit=limit) + result["query"] = query + result["retrieval_query_sha256"] = hashlib.sha256(retrieval_query.encode("utf-8")).hexdigest() + result["retrieval_anchors_sha256"] = hashlib.sha256( + json.dumps(retrieval_anchors, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest() + if getattr(args, "command", None) != "context": + return result + return _enrich_context(args, query, result) + + +def main() -> None: + cloudsql.RUNTIME_SERVICE_ACCOUNT = expected_runtime_service_account() + cloudsql.parse_args = _parse_args + cloudsql.query_rows = _query_rows + cloudsql.main() + + +if __name__ == "__main__": + main() diff --git a/hermes-agent/runtime/leoclean-nosend/profile-sources/leo_db_context.py b/hermes-agent/runtime/leoclean-nosend/profile-sources/leo_db_context.py new file mode 100644 index 0000000..1b6d969 --- /dev/null +++ b/hermes-agent/runtime/leoclean-nosend/profile-sources/leo_db_context.py @@ -0,0 +1,177 @@ +"""GCP-only Cloud SQL adapter over the unchanged shared Leo DB-context plugin.""" + +from __future__ import annotations + +import hashlib +import importlib.util +import json +import os +import subprocess +import sys +from collections.abc import Callable +from datetime import datetime, timezone +from pathlib import Path +from types import ModuleType +from typing import Any + +HERE = Path(__file__).resolve().parent + + +def _load_base() -> ModuleType: + path = HERE / "base.py" + if path.is_symlink() or not path.is_file(): + raise RuntimeError("bound Leo DB-context plugin is unavailable") + spec = importlib.util.spec_from_file_location("livingip_leo_db_context_base", path) + if spec is None or spec.loader is None: + raise RuntimeError("bound Leo DB-context plugin is unavailable") + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +base = _load_base() + + +def _load_database_snapshot( + user_message: str, + *, + conversation_history: Any = None, + hermes_home: Path | None = None, + runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run, +) -> dict[str, Any]: + """Load one private, excerpt-bearing Cloud SQL snapshot without VPS fallback.""" + + query = str(user_message or "")[: base.MAX_QUERY_CHARS] + contextual = getattr(base, "_contextual_retrieval_query", lambda value, _history: (value, [])) + retrieval_query, retrieval_anchors = contextual(query, conversation_history) + resolve_home = getattr( + base, + "_resolve_hermes_home", + lambda explicit: explicit or Path(os.getenv("HERMES_HOME", str(Path.home() / ".hermes"))), + ) + home = resolve_home(hermes_home) + teleo_kb = home / "bin" / "teleo-kb" + query_sha256 = hashlib.sha256(query.encode("utf-8")).hexdigest() + base_record: dict[str, Any] = { + "generated_at_utc": datetime.now(timezone.utc).isoformat(), + "event": "pre_llm_call", + "query_sha256": query_sha256, + "retrieval_query_sha256": hashlib.sha256(retrieval_query.encode("utf-8")).hexdigest(), + "retrieval_anchor_count": len(retrieval_anchors), + "retrieval_anchors_sha256": hashlib.sha256( + json.dumps(retrieval_anchors, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest(), + "source": "teleo-kb context --include-excerpts", + } + if not teleo_kb.is_file() or not os.access(teleo_kb, os.X_OK): + record = base_record | {"status": "error", "error": "teleo_kb_missing", "injected": True} + base._trace(record) + return base._error_snapshot(query, query_sha256, "teleo_kb_missing") + + timeout = int(os.getenv("LEO_DB_CONTEXT_TIMEOUT_SECONDS", str(base.DEFAULT_TIMEOUT_SECONDS))) + command = [ + str(teleo_kb), + "context", + query, + "--retrieval-query", + retrieval_query, + ] + for anchor in retrieval_anchors: + command.extend(("--retrieval-anchor", anchor)) + command.extend( + [ + "--limit", + str(base.CONTEXT_CLAIM_LIMIT), + "--context-limit", + str(base.CONTEXT_ROW_LIMIT), + "--include-excerpts", + "--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} + base._trace(record) + return base._error_snapshot(query, query_sha256, "timeout") + except OSError as exc: + record = base_record | {"status": "error", "error": type(exc).__name__, "injected": True} + base._trace(record) + return base._error_snapshot(query, query_sha256, type(exc).__name__) + + if completed.returncode != 0: + reason = f"teleo_kb_exit_{completed.returncode}" + record = base_record | {"status": "error", "error": reason, "injected": True} + base._trace(record) + return base._error_snapshot(query, query_sha256, reason) + + 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") + claims = payload.get("claims", []) + if not isinstance(claims, list) or not all(isinstance(item, dict) for item in claims): + raise ValueError("claims_missing") + context_rows = payload.get("context_rows", []) + if not isinstance(context_rows, list) or not all(isinstance(item, dict) for item in context_rows): + raise ValueError("context_rows_missing") + compiled_response = payload.get("compiled_response") + if compiled_response is not None and not isinstance(compiled_response, str): + raise ValueError("compiled_response_invalid") + contract_mode = payload.get("runtime_contract_mode") + if contract_mode not in {"claim_evidence_challenge", "general", "review_only_candidate"}: + raise ValueError("runtime_contract_mode_invalid") + if base._requires_database_truth(query) and contract_mode == "general": + record = base_record | { + "status": "error", + "error": "gcp_contract_surface_not_yet_supported", + "injected": True, + } + base._trace(record) + return base._error_snapshot(query, query_sha256, "gcp_contract_surface_not_yet_supported") + retrieval_payload, injected_rows_sha256 = base._compact_retrieval_payload( + claims, + context_rows, + payload.get("retrieval_receipt"), + ) + retrieval_receipt = base._retrieval_receipt_trace( + payload.get("retrieval_receipt"), + injected_rows_sha256=injected_rows_sha256, + ) + except (json.JSONDecodeError, TypeError, ValueError) as exc: + record = base_record | {"status": "error", "error": str(exc), "injected": True} + base._trace(record) + return base._error_snapshot(query, query_sha256, 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(), + "compiled_response_available": bool(compiled_response), + "runtime_contract_mode": contract_mode, + "database_reasoning_protocol_version": base.DATABASE_REASONING_PROTOCOL_VERSION, + } + if retrieval_receipt is not None: + record["retrieval_receipt"] = retrieval_receipt + base._trace(record) + return { + "status": "ok", + "query_sha256": query_sha256, + "contracts": contracts, + "compiled_response": compiled_response, + "requires_database_truth": bool({str(item.get("id") or "") for item in contracts} - {"reply_budget"}), + "database_reasoning_protocol_version": base.DATABASE_REASONING_PROTOCOL_VERSION, + "context": base._format_context(contracts, retrieval_payload, injected_rows_sha256), + } + + +base._load_database_snapshot = _load_database_snapshot + + +def register(ctx: Any) -> None: + base.register(ctx) diff --git a/hermes-agent/runtime/leoclean-nosend/profile-sources/operational_contracts.py b/hermes-agent/runtime/leoclean-nosend/profile-sources/operational_contracts.py new file mode 100644 index 0000000..181a2fa --- /dev/null +++ b/hermes-agent/runtime/leoclean-nosend/profile-sources/operational_contracts.py @@ -0,0 +1,154 @@ +"""SQL-free response contracts for the bounded GCP no-send staging slice.""" + +from __future__ import annotations + +import re +from typing import Any + +SUPPORTED_DATABASE_MODES = frozenset({"claim_evidence_challenge", "review_only_candidate"}) + + +def _reply_budget(query: str) -> dict[str, Any]: + normalized = re.sub(r"[-_/]+", " ", query.lower()) + match = re.search( + r"\b(?:under|within|at most|no more than|maximum(?: of)?|max(?:imum)?(?: of)?)\s+" + r"(?P\d{2,4})\s+words?\b", + normalized, + ) + hard_max = max(20, min(220, int(match.group("words")))) if match else 220 + return { + "id": "reply_budget", + "target_words": min(170, hard_max), + "hard_max_words": hard_max, + "shape": ( + "no intro; at most three compact bullets covering mapping, review/apply, and receipt/limitation; " + "omission is better than exceeding the budget" + ), + } + + +def classify_contract_mode(query: str) -> str: + lowered = query.lower() + existing_claim_audit = any( + term in lowered + for term in ( + "inspect one exact claim", + "exact claim body and its evidence", + "one retrieved canonical claim", + "linked source or evidence id", + ) + ) + candidate = bool( + any( + term in lowered + for term in ( + "candidate claim", + "reviewable candidate", + "candidate only", + "revised exact claim body", + "review-only proposal", + "review only proposal", + ) + ) + and any(term in lowered for term in ("draft", "propose", "revise", "rewrite", "narrow", "return", "turn")) + and not existing_claim_audit + ) + if candidate: + return "review_only_candidate" + challenge = bool( + any(term in lowered for term in ("claim", "exact body", "claim body")) + and any(term in lowered for term in ("evidence", "source")) + and any( + term in lowered + for term in ("challenge", "unsupported", "narrower", "assumption", "what supports", "falsif") + ) + ) + return "claim_evidence_challenge" if challenge else "general" + + +def operational_contracts(query: str) -> list[dict[str, Any]]: + result = [_reply_budget(query)] + if classify_contract_mode(query) == "claim_evidence_challenge": + result.append( + { + "id": "claim_evidence_challenge", + "required_sections": [ + "one retrieved canonical claim ID and exact claim body", + "one linked source or evidence ID and exact evidence excerpt", + "one inference not established by that evidence", + "one narrower evidence-bounded revision", + ], + "boundary": ( + "the claim body and evidence excerpt are distinct database values; do not paraphrase either " + "as if it were the other" + ), + "mutation_policy": "read-only; proposed wording remains a candidate until separately staged and reviewed", + } + ) + return result + + +def _bounded_exact_excerpt(value: Any, *, max_words: int = 45) -> str: + text = str(value or "").strip() + words = list(re.finditer(r"\S+", text)) + if len(words) <= max_words: + return text + return text[: words[max_words - 1].end()].rstrip() + + +def bind_claim_evidence_challenge( + contracts: list[dict[str, Any]], + claims: list[dict[str, Any]], + evidence: dict[str, list[dict[str, Any]]], +) -> None: + contract = next((item for item in contracts if item.get("id") == "claim_evidence_challenge"), None) + if contract is None: + return + for claim in claims: + claim_id = str(claim.get("id") or "") + claim_text = str(claim.get("text") or "").strip() + if not claim_id or not claim_text: + continue + for row in evidence.get(claim_id, []): + source_id = str(row.get("source_id") or "") + excerpt = _bounded_exact_excerpt(row.get("excerpt")) + if not source_id or not excerpt: + continue + contract["binding_status"] = "bound" + contract["selected_claim"] = {"id": claim_id, "text": claim_text} + contract["selected_evidence"] = { + "source_id": source_id, + "excerpt": excerpt, + "role": row.get("role"), + } + return + contract["binding_status"] = "unavailable" + + +def compile_operational_response(contracts: list[dict[str, Any]]) -> str | None: + challenge = next((item for item in contracts if item.get("id") == "claim_evidence_challenge"), None) + if challenge is None: + return None + claim = challenge.get("selected_claim") or {} + evidence = challenge.get("selected_evidence") or {} + claim_id = str(claim.get("id") or "") + claim_text = str(claim.get("text") or "") + source_id = str(evidence.get("source_id") or "") + excerpt = str(evidence.get("excerpt") or "") + if challenge.get("binding_status") != "bound" or not all((claim_id, claim_text, source_id, excerpt)): + return ( + "Live database readback returned no claim/source pair with a non-empty evidence excerpt, so an exact " + "claim-versus-evidence challenge cannot be completed from this result. No identifiers or support were " + "inferred, and no database write was made. Next proof-changing follow-up: broaden the read-only query " + "until one canonical claim and linked source excerpt are returned, then rerun the challenge." + ) + return ( + f"Claim ID: {claim_id}\n" + f'Exact claim body: "{claim_text}"\n\n' + f"Source ID: {source_id}\n" + f'Exact evidence excerpt: "{excerpt}"\n\n' + "Unsupported leap: this evidence excerpt does not establish every broader causal, trend, or generality " + "assertion in the claim body.\n\n" + "Narrower revision: the linked source establishes only the excerpted observation; any broader inference " + "remains unproven by this evidence row alone. No database write was made." + ) diff --git a/hermes-agent/runtime/leoclean-nosend/profile-template/config.yaml b/hermes-agent/runtime/leoclean-nosend/profile-template/config.yaml new file mode 100644 index 0000000..e9b5d8a --- /dev/null +++ b/hermes-agent/runtime/leoclean-nosend/profile-template/config.yaml @@ -0,0 +1,22 @@ +model: + provider: openrouter + default: anthropic/claude-sonnet-4-6 + max_tokens: 8192 +smart_model_routing: false +agent: + max_turns: 90 +memory: + enabled: false + memory_enabled: false + user_profile_enabled: false + nudge_interval: 0 +skills: + creation_nudge_interval: 0 +platform_toolsets: + api_server: + - livingip-leoclean-nosend +known_plugin_toolsets: + api_server: [] +mcp_servers: {} +display: + tool_progress: false diff --git a/hermes-agent/runtime/leoclean-nosend/profile-template/gcp-runtime-identity.json b/hermes-agent/runtime/leoclean-nosend/profile-template/gcp-runtime-identity.json new file mode 100644 index 0000000..1838016 --- /dev/null +++ b/hermes-agent/runtime/leoclean-nosend/profile-template/gcp-runtime-identity.json @@ -0,0 +1,4 @@ +{ + "expected_service_account": "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com", + "schema": "livingip.leocleanGcpRuntimeIdentity.v1" +} diff --git a/hermes-agent/runtime/leoclean-nosend/runtime-contract.json b/hermes-agent/runtime/leoclean-nosend/runtime-contract.json new file mode 100644 index 0000000..383a3ec --- /dev/null +++ b/hermes-agent/runtime/leoclean-nosend/runtime-contract.json @@ -0,0 +1,168 @@ +{ + "schema": "livingip.leocleanNoSendRuntimeContract.v1", + "runtime_mode": "gcp_staging_no_send", + "transport_authority": "absent", + "service_mode": "gateway_without_platform_adapters", + "handler_platform": "api_server", + "plugins": { + "allowed": [ + { + "name": "leo-db-context", + "source": "user", + "tools": [], + "hooks": [ + "post_llm_call", + "pre_llm_call" + ] + } + ] + }, + "toolset": { + "name": "livingip-leoclean-nosend", + "allowed_tools": [ + "skills_list", + "skill_view", + "terminal" + ], + "terminal_commands": [ + "context", + "decision-matrix-status", + "edges", + "evidence", + "list-proposals", + "propose-core-change", + "propose-edge", + "search", + "search-proposals", + "show", + "show-proposal", + "status" + ], + "forbidden_tools": [ + "cronjob", + "delegate_task", + "memory", + "process", + "send_message", + "skill_manage" + ] + }, + "profile": { + "template_entries": [ + "bin", + "config.yaml", + "plugins", + "skills" + ], + "required_runtime_entries": [ + "SOUL.md", + "identity-lock.json", + "identity-manifest.json" + ], + "allowed_runtime_state": { + "directories": [ + "cache", + "cron", + "logs", + "memories", + "platforms", + "sessions", + "state" + ], + "files": [ + "channel_directory.json", + "gateway.pid", + "gateway_state.json", + "state.db", + "state.db-shm", + "state.db-wal" + ] + }, + "forbidden_entries": [ + ".env", + "BOOT.md", + "auth.json", + "gateway.json", + "hooks", + "processes.json", + "secrets" + ], + "learning_policy": { + "automatic_memory_review_interval": 0, + "automatic_skill_review_interval": 0, + "memory_enabled": false, + "user_profile_enabled": false + }, + "routing_policy": { + "cheap_model": "absent", + "smart_model_routing": false + } + }, + "environment": { + "allowed_provider_credentials": [ + "OPENROUTER_API_KEY" + ], + "allowed_runtime_variables": [ + "HERMES_HOME", + "PYTHONDONTWRITEBYTECODE", + "PYTHONNOUSERSITE", + "PYTHONSAFEPATH" + ], + "forbidden_exact": [ + "ALL_PROXY", + "BASH_ENV", + "BOT_TOKEN", + "ENV", + "GATEWAY_ALLOW_ALL_USERS", + "GATEWAY_ALLOWED_USERS", + "GOOGLE_APPLICATION_CREDENTIALS", + "HERMES_ENABLE_PROJECT_PLUGINS", + "HERMES_SESSION_PLATFORM", + "HTTP_PROXY", + "HTTPS_PROXY", + "LD_PRELOAD", + "PYTHONHOME", + "PYTHONPATH", + "TELEGRAM_BOT_TOKEN" + ], + "forbidden_prefixes": [ + "CLOUDSDK_", + "DINGTALK_", + "DISCORD_", + "EMAIL_", + "FEISHU_", + "HERMES_", + "LD_", + "MATRIX_", + "MATTERMOST_", + "SIGNAL_", + "SLACK_", + "SMS_", + "TELEGRAM_", + "TWILIO_", + "WEBCOM_", + "WEBHOOK_", + "WHATSAPP_" + ], + "forbidden_suffixes": [ + "_PASSWORD", + "_SECRET", + "_TOKEN", + "_API_KEY" + ] + }, + "database": { + "canonical_database": "teleo_canonical", + "runtime_role": "leoclean_kb_runtime", + "proposal_function": "kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)", + "direct_canonical_writes": "denied", + "direct_stage_table_writes": "denied", + "proposal_staging": "function_only", + "automatic_context_modes": [ + "claim_evidence_challenge", + "review_only_candidate" + ], + "unsupported_context_policy": "fail_closed" + }, + "claim_ceiling": "This artifact proves reproducible source and a structurally no-send runtime surface. It does not prove live GCP IAM, Cloud SQL identity, model behavior, deployment, restart, parity, Telegram delivery, or production readiness." +} diff --git a/hermes-agent/runtime/leoclean-nosend/runtime-lock.json b/hermes-agent/runtime/leoclean-nosend/runtime-lock.json new file mode 100644 index 0000000..c2115df --- /dev/null +++ b/hermes-agent/runtime/leoclean-nosend/runtime-lock.json @@ -0,0 +1,85 @@ +{ + "schema": "livingip.leocleanNoSendRuntimeLock.v1", + "hermes": { + "repository": "https://github.com/NousResearch/hermes-agent", + "commit": "b2f477a30b3c05d0f383c543af98496ae8a96070", + "archive_url": "https://codeload.github.com/NousResearch/hermes-agent/tar.gz/b2f477a30b3c05d0f383c543af98496ae8a96070", + "archive_sha256": "a1c3455e65d7948746046314c69bf39833d915ca6da75d74e47a89289a36c742", + "archive_root": "hermes-agent-b2f477a30b3c05d0f383c543af98496ae8a96070", + "pyproject_sha256": "da010a5692e65741eb954cf7c00880b999c2c34ec7700ca5508e77cdbc912de5", + "uv_lock_sha256": "636ca22acc1ccfeb1aa022c1b02306726c74a9c36b63fae572fc083b4de9c112", + "patched_source_tree_sha256": "54dafb570844ba0c4d142d484e29ca057dfb323ba753e64f78f7609f0e1f42c4" + }, + "base_images": { + "python": "python:3.11.9-slim-bookworm@sha256:8fb099199b9f2d70342674bd9dbccd3ed03a258f26bbd1d556822c6dfc60c317", + "uv": "ghcr.io/astral-sh/uv:0.9.30@sha256:538e0b39736e7feae937a65983e49d2ab75e1559d35041f9878b7b7e51de91e4" + }, + "patches": [ + { + "source": "hermes-agent/patches/apply_response_transform_hook.py", + "source_sha256": "81b0737fe952639db61873f0fca899d4dc902dfa93cbbe9d4c5d03099258bfd3", + "target": "run_agent.py", + "before_sha256": "d9d1f0bfff5aaed2d92d21cccf9cc95387ce502e406a708f2ccd3874a186cf90", + "after_sha256": "0411cac9a572c002b86588371b4b767203366f176d94dbbcf720cc2e91a36bd3" + }, + { + "source": "hermes-agent/patches/apply_telegram_sender_name_policy.py", + "source_sha256": "aa1470e254068bf88931d7c276f9e4a466b44b2c6f41e6f70b182c754ef39953", + "target": "gateway/platforms/telegram.py", + "before_sha256": "70be41dd47698b9cbb3f28d3f29a73074f93b39fbef7e0f152a8479abf62f181", + "after_sha256": "a5aebb97bff6ef8f0d3801156de047d1d03000dfd03e5adadd89224f36129102" + } + ], + "profile_sources": [ + { + "source": "hermes-agent/runtime/leoclean-nosend/profile-template/config.yaml", + "target": "config.yaml", + "mode": "0600" + }, + { + "source": "hermes-agent/leoclean-bin/teleo-kb-gcp", + "target": "bin/teleo-kb", + "mode": "0755" + }, + { + "source": "hermes-agent/leoclean-bin/cloudsql_memory_tool.py", + "target": "bin/cloudsql_memory_base.py", + "mode": "0644" + }, + { + "source": "hermes-agent/runtime/leoclean-nosend/profile-sources/operational_contracts.py", + "target": "bin/operational_contracts.py", + "mode": "0644" + }, + { + "source": "hermes-agent/runtime/leoclean-nosend/profile-sources/cloudsql_memory_runtime.py", + "target": "bin/cloudsql_memory_tool.py", + "mode": "0755" + }, + { + "source": "hermes-agent/runtime/leoclean-nosend/profile-template/gcp-runtime-identity.json", + "target": "bin/gcp-runtime-identity.json", + "mode": "0644" + }, + { + "source": "hermes-agent/leoclean-skills/gcp/teleo-kb-bridge/SKILL.md", + "target": "skills/teleo-kb-bridge/SKILL.md", + "mode": "0644" + }, + { + "source": "hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py", + "target": "plugins/leo-db-context/base.py", + "mode": "0644" + }, + { + "source": "hermes-agent/runtime/leoclean-nosend/profile-sources/leo_db_context.py", + "target": "plugins/leo-db-context/__init__.py", + "mode": "0644" + }, + { + "source": "hermes-agent/leoclean-plugins/vps/leo-db-context/plugin.yaml", + "target": "plugins/leo-db-context/plugin.yaml", + "mode": "0644" + } + ] +} diff --git a/scripts/compile_leoclean_nosend_runtime.py b/scripts/compile_leoclean_nosend_runtime.py new file mode 100644 index 0000000..48b0237 --- /dev/null +++ b/scripts/compile_leoclean_nosend_runtime.py @@ -0,0 +1,381 @@ +#!/usr/bin/env python3 +"""Assemble the source-pinned, secret-free Hermes/leoclean no-send artifact.""" + +from __future__ import annotations + +import argparse +import hashlib +import importlib.util +import json +import os +import shutil +import stat +import subprocess +import tarfile +import tempfile +import urllib.error +import urllib.request +from pathlib import Path, PurePosixPath +from typing import Any + +LOCK_SCHEMA = "livingip.leocleanNoSendRuntimeLock.v1" +MANIFEST_SCHEMA = "livingip.leocleanNoSendArtifactManifest.v1" +RUNTIME_REL = Path("hermes-agent/runtime/leoclean-nosend") + + +class CompileError(RuntimeError): + """The artifact could not be assembled without weakening its contract.""" + + +def _require(condition: bool, message: str) -> None: + if not condition: + raise CompileError(message) + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _canonical_sha256(value: Any) -> str: + encoded = json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _is_sha256(value: Any) -> bool: + return ( + isinstance(value, str) + and len(value) == 64 + and all(character in "0123456789abcdef" for character in value) + ) + + +def _safe_relative(value: Any, label: str) -> str: + _require(isinstance(value, str) and bool(value), f"{label} must be a non-empty relative path") + _require("\\" not in value and "\x00" not in value, f"{label} contains an unsafe character") + path = PurePosixPath(value) + _require(not path.is_absolute() and "." not in path.parts and ".." not in path.parts, f"{label} is unsafe") + return path.as_posix() + + +def _load_json(path: Path, label: str) -> dict[str, Any]: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise CompileError(f"cannot read {label}: {type(exc).__name__}") from exc + _require(isinstance(value, dict), f"{label} must be a JSON object") + return value + + +def _validate_lock(value: dict[str, Any]) -> None: + _require(value.get("schema") == LOCK_SCHEMA, "unsupported runtime lock schema") + _require(set(value) == {"schema", "hermes", "base_images", "patches", "profile_sources"}, "runtime lock fields are invalid") + hermes = value.get("hermes") + _require(isinstance(hermes, dict), "Hermes lock is missing") + _require( + set(hermes) + == { + "repository", + "commit", + "archive_url", + "archive_sha256", + "archive_root", + "pyproject_sha256", + "uv_lock_sha256", + "patched_source_tree_sha256", + }, + "Hermes lock fields are invalid", + ) + commit = hermes.get("commit") + _require(isinstance(commit, str) and len(commit) == 40 and all(ch in "0123456789abcdef" for ch in commit), "Hermes commit must be an exact SHA") + _require(hermes.get("repository") == "https://github.com/NousResearch/hermes-agent", "Hermes repository is invalid") + _require( + hermes.get("archive_url") == f"https://codeload.github.com/NousResearch/hermes-agent/tar.gz/{commit}", + "Hermes archive URL is not commit-bound", + ) + _require(hermes.get("archive_root") == f"hermes-agent-{commit}", "Hermes archive root is invalid") + for field in ("archive_sha256", "pyproject_sha256", "uv_lock_sha256", "patched_source_tree_sha256"): + _require(_is_sha256(hermes.get(field)), f"Hermes {field} is invalid") + images = value.get("base_images") + _require(isinstance(images, dict) and set(images) == {"python", "uv"}, "base image lock is invalid") + for image in images.values(): + _require(isinstance(image, str) and "@sha256:" in image, "base images must use digest references") + patches = value.get("patches") + _require(isinstance(patches, list) and len(patches) == 2, "exactly two managed patches are required") + targets: set[str] = set() + for patch in patches: + _require(isinstance(patch, dict), "patch lock entry must be an object") + _require(set(patch) == {"source", "source_sha256", "target", "before_sha256", "after_sha256"}, "patch lock fields are invalid") + _safe_relative(patch["source"], "patch source") + target = _safe_relative(patch["target"], "patch target") + _require(isinstance(target, str) and target not in targets, "patch targets must be unique") + targets.add(target) + for field in ("source_sha256", "before_sha256", "after_sha256"): + _require(_is_sha256(patch[field]), f"patch {field} is invalid") + sources = value.get("profile_sources") + _require(isinstance(sources, list) and bool(sources), "profile sources are missing") + profile_targets: set[str] = set() + for source in sources: + _require(isinstance(source, dict) and set(source) == {"source", "target", "mode"}, "profile source fields are invalid") + _safe_relative(source["source"], "profile source") + target = _safe_relative(source["target"], "profile target") + _require(target not in profile_targets, "duplicate profile target") + profile_targets.add(target) + _require(source["mode"] in {"0600", "0644", "0755"}, "profile mode is invalid") + + +def _download(url: str, target: Path, timeout: float) -> None: + request = urllib.request.Request(url, headers={"User-Agent": "livingip-runtime-compiler/1"}) + try: + with urllib.request.urlopen(request, timeout=timeout) as response, target.open("wb") as handle: + _require(response.status == 200, f"Hermes archive returned HTTP {response.status}") + while chunk := response.read(1024 * 1024): + handle.write(chunk) + except (OSError, urllib.error.URLError) as exc: + raise CompileError(f"cannot download Hermes archive: {type(exc).__name__}") from exc + + +def _safe_extract(archive: Path, destination: Path, expected_root: str) -> Path: + try: + with tarfile.open(archive, "r:gz") as bundle: + members = bundle.getmembers() + _require(bool(members), "Hermes archive is empty") + for member in members: + path = PurePosixPath(member.name) + _require(not path.is_absolute() and ".." not in path.parts, "Hermes archive contains an unsafe path") + _require(path.parts and path.parts[0] == expected_root, "Hermes archive root is unexpected") + _require(not member.issym() and not member.islnk(), "Hermes archive contains a link") + _require(member.isfile() or member.isdir(), "Hermes archive contains an unsupported entry") + bundle.extractall(destination, members=members) + except (OSError, tarfile.TarError) as exc: + raise CompileError(f"cannot extract Hermes archive: {type(exc).__name__}") from exc + root = destination / expected_root + _require(root.is_dir() and not root.is_symlink(), "extracted Hermes root is missing or unsafe") + return root + + +def _bound_file(root: Path, relative: str, label: str) -> Path: + """Return one regular file whose full path stays inside *root*.""" + + root_resolved = root.resolve(strict=True) + candidate = root / _safe_relative(relative, label) + try: + resolved = candidate.resolve(strict=True) + except OSError as exc: + raise CompileError(f"{label} is missing or unsafe") from exc + _require(resolved.is_relative_to(root_resolved), f"{label} escapes its declared root") + current = candidate + while current != root: + _require(not current.is_symlink(), f"{label} traverses a symlink") + current = current.parent + _require(resolved.is_file(), f"{label} is not a regular file") + return resolved + + +def _load_patcher(path: Path): + spec = importlib.util.spec_from_file_location(f"livingip_runtime_patch_{path.stem}", path) + _require(spec is not None and spec.loader is not None, f"cannot load managed patcher: {path}") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + _require(callable(getattr(module, "install", None)), f"managed patcher has no install function: {path}") + return module + + +def _apply_patches(repo_root: Path, hermes_root: Path, lock: dict[str, Any]) -> list[dict[str, str]]: + receipts: list[dict[str, str]] = [] + for entry in lock["patches"]: + patcher = _bound_file(repo_root, entry["source"], f"managed patcher {entry['source']}") + target = _bound_file(hermes_root, entry["target"], f"Hermes patch target {entry['target']}") + _require(_sha256(patcher) == entry["source_sha256"], f"managed patcher drifted: {entry['source']}") + _require(_sha256(target) == entry["before_sha256"], f"Hermes patch target drifted before apply: {entry['target']}") + module = _load_patcher(patcher) + result, code = module.install(target, check=False) + _require(code == 0, f"managed patch failed: {entry['source']}") + _require(_sha256(target) == entry["after_sha256"], f"Hermes patch result drifted: {entry['target']}") + _result, check_code = module.install(target, check=True) + _require(check_code == 0, f"managed patch is not idempotent: {entry['source']}") + receipts.append( + { + "source": entry["source"], + "source_sha256": entry["source_sha256"], + "target": entry["target"], + "before_sha256": entry["before_sha256"], + "after_sha256": entry["after_sha256"], + "status": str(result.get("status", "installed")), + } + ) + return receipts + + +def _copy_profile(repo_root: Path, artifact: Path, lock: dict[str, Any]) -> None: + profile = artifact / "profile-template" + profile.mkdir(mode=0o700) + for entry in lock["profile_sources"]: + source = _bound_file(repo_root, entry["source"], f"profile source {entry['source']}") + target = profile / entry["target"] + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copyfile(source, target) + target.chmod(int(entry["mode"], 8)) + + +def _tree_manifest(root: Path, *, excluded: frozenset[str] = frozenset()) -> dict[str, Any]: + files: list[dict[str, Any]] = [] + for path in sorted(root.rglob("*"), key=lambda item: item.relative_to(root).as_posix()): + relative = path.relative_to(root).as_posix() + if relative in excluded: + continue + _require(not path.is_symlink(), f"artifact contains a symlink: {relative}") + if path.is_dir(): + continue + _require(path.is_file(), f"artifact contains an unsupported entry: {relative}") + mode = stat.S_IMODE(path.stat().st_mode) + files.append( + { + "path": relative, + "bytes": path.stat().st_size, + "mode": f"0o{mode:03o}", + "sha256": _sha256(path), + } + ) + stable = {"files": files, "file_count": len(files), "total_bytes": sum(item["bytes"] for item in files)} + return {**stable, "sha256": _canonical_sha256(stable)} + + +def _git_state(repo_root: Path, relevant_paths: list[str]) -> tuple[str, bool]: + try: + head = subprocess.run( + ["git", "-C", str(repo_root), "rev-parse", "HEAD"], + check=True, + text=True, + capture_output=True, + ).stdout.strip() + status = subprocess.run( + ["git", "-C", str(repo_root), "status", "--porcelain", "--", *relevant_paths], + check=True, + text=True, + capture_output=True, + ).stdout.strip() + except (OSError, subprocess.CalledProcessError) as exc: + raise CompileError(f"cannot inspect Teleo source revision: {type(exc).__name__}") from exc + return head, bool(status) + + +def compile_artifact( + *, + repo_root: Path, + output: Path, + archive: Path | None, + allow_dirty: bool, + timeout: float, +) -> dict[str, Any]: + runtime_root = repo_root / RUNTIME_REL + lock_path = runtime_root / "runtime-lock.json" + lock = _load_json(lock_path, "runtime lock") + _validate_lock(lock) + relevant = [ + str(RUNTIME_REL / name) + for name in ("bootstrap.py", "profile-template/config.yaml", "runtime-contract.json", "runtime-lock.json") + ] + relevant.extend(entry["source"] for entry in lock["patches"]) + relevant.extend(entry["source"] for entry in lock["profile_sources"]) + relevant.extend( + [ + "scripts/compile_leoclean_nosend_runtime.py", + "scripts/run_leoclean_nosend_runtime_canary.sh", + "scripts/verify_leoclean_nosend_runtime.py", + ] + ) + teleo_head, dirty = _git_state(repo_root, sorted(set(relevant))) + _require(allow_dirty or not dirty, "refusing to compile from dirty or untracked runtime sources") + _require(not output.exists(), "artifact output already exists") + + output.parent.mkdir(parents=True, exist_ok=True) + staging = Path(tempfile.mkdtemp(prefix=f".{output.name}.", dir=output.parent)) + try: + archive_path = archive + if archive_path is None: + archive_path = staging / "hermes.tar.gz" + _download(lock["hermes"]["archive_url"], archive_path, timeout) + _require(archive_path.is_file() and not archive_path.is_symlink(), "Hermes archive is missing or unsafe") + _require(_sha256(archive_path) == lock["hermes"]["archive_sha256"], "Hermes archive SHA-256 mismatch") + extracted = staging / "extracted" + extracted.mkdir() + hermes_source = _safe_extract(archive_path, extracted, lock["hermes"]["archive_root"]) + _require(_sha256(hermes_source / "pyproject.toml") == lock["hermes"]["pyproject_sha256"], "Hermes pyproject drifted") + _require(_sha256(hermes_source / "uv.lock") == lock["hermes"]["uv_lock_sha256"], "Hermes dependency lock drifted") + patch_receipts = _apply_patches(repo_root, hermes_source, lock) + patched_source = _tree_manifest(hermes_source) + _require( + patched_source["sha256"] == lock["hermes"]["patched_source_tree_sha256"], + "patched Hermes source tree drifted", + ) + + artifact = staging / "artifact" + artifact.mkdir(mode=0o755) + shutil.copytree(hermes_source, artifact / "hermes-agent", symlinks=False) + _copy_profile(repo_root, artifact, lock) + for name in ("bootstrap.py", "runtime-contract.json", "runtime-lock.json"): + source = runtime_root / name + _require(source.is_file() and not source.is_symlink(), f"runtime source is missing or unsafe: {name}") + shutil.copyfile(source, artifact / name) + (artifact / "bootstrap.py").chmod(0o755) + (artifact / "runtime-contract.json").chmod(0o644) + (artifact / "runtime-lock.json").chmod(0o644) + + content = _tree_manifest(artifact) + manifest = { + "schema": MANIFEST_SCHEMA, + "status": "pass", + "teleo_git_head": teleo_head, + "teleo_relevant_sources_clean": not dirty, + "hermes_commit": lock["hermes"]["commit"], + "hermes_archive_sha256": lock["hermes"]["archive_sha256"], + "dependency_lock_sha256": lock["hermes"]["uv_lock_sha256"], + "hermes_source_tree": patched_source, + "patches": patch_receipts, + "content": content, + "claim_ceiling": _load_json(runtime_root / "runtime-contract.json", "runtime contract")["claim_ceiling"], + } + (artifact / "artifact-manifest.json").write_text( + json.dumps(manifest, sort_keys=True, indent=2) + "\n", + encoding="utf-8", + ) + (artifact / "artifact-manifest.json").chmod(0o644) + os.replace(artifact, output) + return manifest + except Exception: + shutil.rmtree(staging, ignore_errors=True) + raise + finally: + if staging.exists(): + shutil.rmtree(staging, ignore_errors=True) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--archive", type=Path) + parser.add_argument("--allow-dirty", action="store_true") + parser.add_argument("--download-timeout", type=float, default=30.0) + args = parser.parse_args() + try: + result = compile_artifact( + repo_root=args.repo_root.resolve(), + output=args.output.resolve(), + archive=args.archive.resolve() if args.archive else None, + allow_dirty=args.allow_dirty, + timeout=args.download_timeout, + ) + except CompileError as exc: + print(json.dumps({"status": "fail", "error": str(exc)}, sort_keys=True)) + return 65 + print(json.dumps(result, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/run_leoclean_nosend_runtime_canary.sh b/scripts/run_leoclean_nosend_runtime_canary.sh new file mode 100755 index 0000000..34501ac --- /dev/null +++ b/scripts/run_leoclean_nosend_runtime_canary.sh @@ -0,0 +1,84 @@ +#!/usr/bin/env bash +set -euo pipefail + +repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +results_dir="${LEOCLEAN_NOSEND_RESULTS_DIR:-${repo_root}/.crabbox-results}" +receipt="${LEOCLEAN_NOSEND_RECEIPT:-${results_dir}/leoclean-nosend-runtime.json}" +uv_bin="${UV_BIN:-$(command -v uv || true)}" +python_bin="${PYTHON_BIN:-python3}" + +if [[ -z "$uv_bin" || ! -x "$uv_bin" ]]; then + echo "uv 0.9.30 is required" >&2 + exit 69 +fi +if [[ "$($uv_bin --version | awk '{print $2}')" != "0.9.30" ]]; then + echo "uv must be exactly 0.9.30" >&2 + exit 69 +fi +if ! command -v "$python_bin" >/dev/null 2>&1; then + echo "the compiler Python is unavailable: $python_bin" >&2 + exit 69 +fi + +temporary="$(mktemp -d "${TMPDIR:-/tmp}/livingip-nosend-runtime.XXXXXX")" +cleanup() { + rm -rf "$temporary" +} +trap cleanup EXIT INT TERM + +artifact="$temporary/artifact" +repeat_artifact="$temporary/artifact-repeat" +hermes_build="$temporary/hermes-build" +runtime_venv="$temporary/runtime-venv" +mkdir -p "$results_dir" "$(dirname "$receipt")" +rm -f "$receipt" + +"$python_bin" "$repo_root/scripts/compile_leoclean_nosend_runtime.py" \ + --repo-root "$repo_root" \ + --output "$artifact" \ + --download-timeout 60 \ + >"$temporary/compile-primary.json" +"$python_bin" "$repo_root/scripts/compile_leoclean_nosend_runtime.py" \ + --repo-root "$repo_root" \ + --output "$repeat_artifact" \ + --download-timeout 60 \ + >"$temporary/compile-repeat.json" + +"$python_bin" - "$artifact/artifact-manifest.json" "$repeat_artifact/artifact-manifest.json" <<'PY' +import json +import sys +from pathlib import Path + +first = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8")) +second = json.loads(Path(sys.argv[2]).read_text(encoding="utf-8")) +if first != second or first.get("content", {}).get("sha256") != second.get("content", {}).get("sha256"): + raise SystemExit("two clean compiles did not produce the same artifact manifest") +PY + +cp -R "$artifact/hermes-agent" "$hermes_build" +"$uv_bin" venv --python 3.11.9 --no-project "$runtime_venv" +VIRTUAL_ENV="$runtime_venv" "$uv_bin" sync \ + --directory "$hermes_build" \ + --active \ + --frozen \ + --no-dev \ + --extra cron \ + --no-editable + +"$runtime_venv/bin/python" "$repo_root/scripts/verify_leoclean_nosend_runtime.py" \ + --repo-root "$repo_root" \ + --artifact "$artifact" \ + --runtime-python "$runtime_venv/bin/python" \ + --uv-bin "$uv_bin" \ + --output "$receipt" + +"$runtime_venv/bin/python" - "$receipt" <<'PY' +import json +import sys +from pathlib import Path + +receipt = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8")) +if receipt.get("status") != "pass" or receipt.get("verification_mode") != "release": + raise SystemExit("no-send runtime receipt did not pass in release mode") +print(json.dumps({"artifact_sha256": receipt["artifact_sha256"], "status": "pass"}, sort_keys=True)) +PY diff --git a/scripts/verify_leoclean_nosend_runtime.py b/scripts/verify_leoclean_nosend_runtime.py new file mode 100644 index 0000000..62a41ef --- /dev/null +++ b/scripts/verify_leoclean_nosend_runtime.py @@ -0,0 +1,417 @@ +#!/usr/bin/env python3 +"""Verify a compiled Hermes/leoclean artifact against the committed no-send lock.""" + +from __future__ import annotations + +import argparse +import json +import os +import secrets +import shutil +import stat +import subprocess +import sys +import tempfile +from pathlib import Path +from typing import Any + +import yaml + +if __package__ in {None, ""}: + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from scripts import compile_leoclean_nosend_runtime as compiler + +RECEIPT_SCHEMA = "livingip.leocleanNoSendArtifactVerification.v1" + + +class VerificationError(RuntimeError): + """The artifact does not match its committed source and no-send contract.""" + + +def _require(condition: bool, message: str) -> None: + if not condition: + raise VerificationError(message) + + +def _load_json(path: Path, label: str) -> dict[str, Any]: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise VerificationError(f"cannot read {label}: {type(exc).__name__}") from exc + _require(isinstance(value, dict), f"{label} must be a JSON object") + return value + + +def _validate_profile_config(profile: Path, contract: dict[str, Any]) -> dict[str, Any]: + try: + value = yaml.safe_load((profile / "config.yaml").read_text(encoding="utf-8")) or {} + except (OSError, TypeError, yaml.YAMLError) as exc: + raise VerificationError(f"cannot read profile config: {type(exc).__name__}") from exc + _require(isinstance(value, dict), "profile config must be a mapping") + _require( + set(value) + == { + "agent", + "display", + "known_plugin_toolsets", + "mcp_servers", + "memory", + "model", + "platform_toolsets", + "skills", + "smart_model_routing", + }, + "profile config top-level fields are not exact", + ) + toolset = contract["toolset"]["name"] + _require(value.get("platform_toolsets") == {"api_server": [toolset]}, "profile toolset binding is not exact") + _require(value.get("known_plugin_toolsets") == {"api_server": []}, "plugin toolset allowlist is not exact") + _require(value.get("mcp_servers") == {}, "profile enables an MCP server") + memory = value.get("memory") or {} + _require( + set(memory) == {"enabled", "memory_enabled", "nudge_interval", "user_profile_enabled"} + and memory.get("enabled") is False + and memory.get("memory_enabled") is False + and memory.get("user_profile_enabled") is False + and type(memory.get("nudge_interval")) is int + and memory.get("nudge_interval") == 0, + "profile enables persistent memory or automatic memory review", + ) + skills = value.get("skills") or {} + _require( + set(skills) == {"creation_nudge_interval"} + and type(skills.get("creation_nudge_interval")) is int + and skills.get("creation_nudge_interval") == 0, + "profile enables automatic skill review", + ) + _require(value.get("smart_model_routing") is False, "profile enables smart-model routing") + encoded = json.dumps(value, sort_keys=True).lower() + forbidden = ( + "telegram", + "discord", + "whatsapp", + "slack", + "signal", + "mattermost", + "matrix", + "dingtalk", + "feishu", + "wecom", + "webhook", + "bot_token", + "send_message", + ) + for marker in forbidden: + _require(marker not in encoded, f"profile config contains forbidden transport marker: {marker}") + return value + + +def _verify_source_bindings(repo_root: Path, artifact: Path, lock: dict[str, Any]) -> dict[str, str]: + bindings: dict[str, str] = {} + for name in ("bootstrap.py", "runtime-contract.json", "runtime-lock.json"): + expected = repo_root / compiler.RUNTIME_REL / name + observed = artifact / name + _require(expected.is_file() and observed.is_file(), f"runtime binding is missing: {name}") + _require(compiler._sha256(expected) == compiler._sha256(observed), f"runtime binding drifted: {name}") + expected_mode = 0o755 if name == "bootstrap.py" else 0o644 + _require(stat.S_IMODE(observed.stat().st_mode) == expected_mode, f"runtime binding mode drifted: {name}") + bindings[name] = compiler._sha256(observed) + for entry in lock["profile_sources"]: + expected = repo_root / entry["source"] + observed = artifact / "profile-template" / entry["target"] + _require(expected.is_file() and observed.is_file(), f"profile binding is missing: {entry['target']}") + _require( + compiler._sha256(expected) == compiler._sha256(observed), f"profile binding drifted: {entry['target']}" + ) + _require( + stat.S_IMODE(observed.stat().st_mode) == int(entry["mode"], 8), + f"profile binding mode drifted: {entry['target']}", + ) + bindings[f"profile-template/{entry['target']}"] = compiler._sha256(observed) + return bindings + + +def _verify_dependency_lock(artifact: Path, uv_bin: str | None, *, required: bool) -> dict[str, Any]: + executable = uv_bin or shutil.which("uv") + if not executable: + _require(not required, "uv is required for release verification") + return {"status": "not_run", "reason": "uv_unavailable"} + environment = { + "HOME": os.environ.get("HOME", "/tmp"), + "PATH": os.environ.get("PATH", ""), + "UV_NO_CACHE": "1", + } + result = subprocess.run( + [executable, "lock", "--check", "--directory", str(artifact / "hermes-agent")], + text=True, + capture_output=True, + check=False, + env=environment, + timeout=60, + ) + _require(result.returncode == 0, "Hermes dependency lock is not current") + return {"status": "pass", "command": "uv lock --check", "stdout": result.stdout.strip()} + + +def _run_runtime_probe(artifact: Path, python: Path) -> dict[str, Any]: + _require(python.is_file() and os.access(python, os.X_OK), "runtime Python is missing or not executable") + with tempfile.TemporaryDirectory(prefix="livingip-nosend-probe-") as temporary: + temporary_root = Path(temporary) + profile = temporary_root / "profile" + shutil.copytree(artifact / "profile-template", profile, symlinks=False) + home = temporary_root / "home" + home.mkdir() + credential_sentinel = f"livingip-probe-{secrets.token_hex(24)}" + environment = { + "HOME": str(home), + "HERMES_HOME": str(profile), + "LANG": "C", + "LC_ALL": "C", + "PATH": os.environ.get("PATH", ""), + "PYTHONDONTWRITEBYTECODE": "1", + "PYTHONNOUSERSITE": "1", + "PYTHONSAFEPATH": "1", + "OPENROUTER_API_KEY": credential_sentinel, + } + result = subprocess.run( + [ + str(python), + str(artifact / "bootstrap.py"), + "probe", + "--profile", + str(profile), + "--template", + ], + text=True, + capture_output=True, + check=False, + env=environment, + timeout=90, + ) + _require(credential_sentinel not in result.stdout, "runtime probe emitted the provider credential sentinel") + _require(credential_sentinel not in result.stderr, "runtime probe emitted the provider credential sentinel") + _require(result.returncode == 0, f"effective runtime probe failed: {result.stderr.strip()[:4000]}") + try: + value = json.loads(result.stdout) + except json.JSONDecodeError as exc: + raise VerificationError("effective runtime probe did not emit JSON") from exc + _require(value.get("status") == "pass", "effective runtime probe did not pass") + _require(value.get("gateway_adapter_count") == 0, "effective runtime has a gateway adapter") + _require(value.get("connected_platforms") == [], "effective runtime has a connected platform") + _require(value.get("python_version") == "3.11.9", "effective runtime Python is not exactly 3.11.9") + surface = value.get("tool_surface") or {} + _require(surface.get("send_message_present") is False, "send_message is present in the effective runtime") + _require(surface.get("tools") == ["skill_view", "skills_list", "terminal"], "effective tool registry is not exact") + _require(surface.get("registry_sealed") is True, "effective tool registry is not sealed") + _require(surface.get("terminal_handler") == "livingip.restricted_teleo_kb.v1", "terminal handler is not restricted") + _require( + surface.get("terminal_restricted_to") == "profile/bin/teleo-kb", + "terminal target receipt is not portable", + ) + _require( + value.get("provider_credential_names_present") == ["OPENROUTER_API_KEY"], + "runtime probe did not observe exactly the synthetic provider credential", + ) + _require( + "provider_credential_values_retained" not in value, + "runtime probe made an unsupported credential-retention claim", + ) + _require( + value.get("effective_policy") + == { + "cheap_model_route": None, + "memory_enabled": False, + "memory_nudge_interval": 0, + "skill_creation_nudge_interval": 0, + "smart_model_routing": False, + "user_profile_enabled": False, + }, + "effective Hermes learning or routing policy is not disabled", + ) + lifecycle = value.get("lifecycle") or {} + _require( + lifecycle.get("started") is True and lifecycle.get("stopped") is True, "runtime lifecycle was not exercised" + ) + for phase in ("post_start_registry", "post_discovery_registry"): + observed = lifecycle.get(phase) or {} + _require(observed.get("tools") == ["skill_view", "skills_list", "terminal"], f"{phase} is not exact") + _require(observed.get("registry_sealed") is True, f"{phase} is not sealed") + _require(observed.get("terminal_restricted_to") == "profile/bin/teleo-kb", f"{phase} target is not portable") + _require( + lifecycle.get("gateway_hooks") == [{"events": ["gateway:startup"], "name": "boot-md", "path": "(builtin)"}], + "gateway hook inventory is not exact", + ) + value["provider_credential_sentinel_redaction"] = "pass" + return value + + +def _git_head(repo_root: Path) -> str: + try: + return subprocess.run( + ["git", "-C", str(repo_root), "rev-parse", "HEAD"], + check=True, + text=True, + capture_output=True, + ).stdout.strip() + except (OSError, subprocess.CalledProcessError) as exc: + raise VerificationError(f"cannot inspect Teleo source revision: {type(exc).__name__}") from exc + + +def verify_artifact( + *, + repo_root: Path, + artifact: Path, + runtime_python: Path | None, + uv_bin: str | None, + release: bool, +) -> dict[str, Any]: + _require(artifact.is_dir() and not artifact.is_symlink(), "artifact root is missing or unsafe") + manifest = _load_json(artifact / "artifact-manifest.json", "artifact manifest") + _require(manifest.get("schema") == compiler.MANIFEST_SCHEMA, "unsupported artifact manifest schema") + _require(manifest.get("status") == "pass", "artifact manifest status is not pass") + _require( + set(manifest) + == { + "schema", + "status", + "teleo_git_head", + "teleo_relevant_sources_clean", + "hermes_commit", + "hermes_archive_sha256", + "dependency_lock_sha256", + "hermes_source_tree", + "patches", + "content", + "claim_ceiling", + }, + "artifact manifest fields are not exact", + ) + expected_lock = _load_json(repo_root / compiler.RUNTIME_REL / "runtime-lock.json", "committed runtime lock") + compiler._validate_lock(expected_lock) + artifact_lock = _load_json(artifact / "runtime-lock.json", "artifact runtime lock") + _require(artifact_lock == expected_lock, "artifact runtime lock differs from the committed lock") + contract = _load_json(artifact / "runtime-contract.json", "runtime contract") + _require( + contract.get("schema") == "livingip.leocleanNoSendRuntimeContract.v1", "runtime contract schema is invalid" + ) + + content = compiler._tree_manifest(artifact, excluded=frozenset({"artifact-manifest.json"})) + _require(content == manifest.get("content"), "artifact content manifest does not match the files") + _require( + manifest.get("teleo_git_head") == _git_head(repo_root), "artifact Teleo revision is not the current revision" + ) + if release: + _require( + manifest.get("teleo_relevant_sources_clean") is True, "release artifact was compiled from dirty sources" + ) + _require(manifest.get("hermes_commit") == expected_lock["hermes"]["commit"], "artifact Hermes commit drifted") + _require( + manifest.get("hermes_archive_sha256") == expected_lock["hermes"]["archive_sha256"], + "artifact Hermes archive binding drifted", + ) + _require( + manifest.get("dependency_lock_sha256") == expected_lock["hermes"]["uv_lock_sha256"], + "artifact dependency lock binding drifted", + ) + _require(manifest.get("claim_ceiling") == contract.get("claim_ceiling"), "artifact claim ceiling drifted") + expected_patches = [ + { + **patch, + "status": "installed_now", + } + for patch in expected_lock["patches"] + ] + _require(manifest.get("patches") == expected_patches, "artifact patch receipts drifted") + hermes_tree = compiler._tree_manifest(artifact / "hermes-agent") + _require( + hermes_tree.get("sha256") == expected_lock["hermes"]["patched_source_tree_sha256"], + "patched Hermes source tree does not match the lock", + ) + _require(hermes_tree == manifest.get("hermes_source_tree"), "Hermes source tree manifest is inconsistent") + _require( + compiler._sha256(artifact / "hermes-agent" / "pyproject.toml") == expected_lock["hermes"]["pyproject_sha256"], + "Hermes pyproject does not match the lock", + ) + _require( + compiler._sha256(artifact / "hermes-agent" / "uv.lock") == expected_lock["hermes"]["uv_lock_sha256"], + "Hermes dependency lock does not match the lock", + ) + for patch in expected_lock["patches"]: + target = artifact / "hermes-agent" / patch["target"] + _require(compiler._sha256(target) == patch["after_sha256"], f"patched target does not match: {patch['target']}") + + profile = artifact / "profile-template" + _require(not any(path.is_symlink() for path in artifact.rglob("*")), "artifact contains a symlink") + actual_entries = {path.name for path in profile.iterdir()} + _require(actual_entries == set(contract["profile"]["template_entries"]), "profile template entry set is not exact") + _require(not (actual_entries & set(contract["profile"]["forbidden_entries"])), "profile contains a forbidden entry") + config = _validate_profile_config(profile, contract) + allowed_tools = contract["toolset"]["allowed_tools"] + _require("send_message" not in allowed_tools, "no-send contract allows send_message") + _require("process" not in allowed_tools, "no-send contract allows the process tool") + + source_bindings = _verify_source_bindings(repo_root, artifact, expected_lock) + dependency_check = _verify_dependency_lock(artifact, uv_bin, required=release) + if runtime_python: + runtime_probe = _run_runtime_probe(artifact, runtime_python) + else: + _require(not release, "runtime Python is required for release verification") + runtime_probe = {"status": "not_run", "reason": "runtime_python_not_supplied"} + _require( + compiler._tree_manifest(artifact, excluded=frozenset({"artifact-manifest.json"})) == content, + "runtime probe mutated the artifact", + ) + return { + "schema": RECEIPT_SCHEMA, + "status": "pass" if release else "incomplete", + "verification_mode": "release" if release else "development", + "artifact_sha256": content["sha256"], + "teleo_git_head": manifest.get("teleo_git_head"), + "hermes_commit": manifest.get("hermes_commit"), + "hermes_source_sha256": hermes_tree["sha256"], + "config_sha256": compiler._sha256(profile / "config.yaml"), + "source_bindings": source_bindings, + "dependency_lock_check": dependency_check, + "runtime_probe": runtime_probe, + "declared_toolset": config["platform_toolsets"]["api_server"], + "claim_ceiling": contract["claim_ceiling"], + } + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) + parser.add_argument("--artifact", type=Path, required=True) + parser.add_argument("--runtime-python", type=Path) + parser.add_argument("--uv-bin") + parser.add_argument("--output", type=Path) + parser.add_argument( + "--development", + action="store_true", + help="Allow dirty sources or omitted executable checks, but never emit a release pass.", + ) + args = parser.parse_args() + try: + receipt = verify_artifact( + repo_root=args.repo_root.resolve(), + artifact=args.artifact.resolve(), + # A virtual-environment interpreter is commonly a symlink to the + # base Python binary. Resolving it would discard the virtual + # environment's prefix and probe the wrong dependency set. + runtime_python=Path(os.path.abspath(args.runtime_python)) if args.runtime_python else None, + uv_bin=args.uv_bin, + release=not args.development, + ) + except (VerificationError, compiler.CompileError) as exc: + print(json.dumps({"status": "fail", "error": str(exc)}, sort_keys=True)) + return 65 + encoded = json.dumps(receipt, sort_keys=True, indent=2) + "\n" + if args.output: + args.output.write_text(encoded, encoding="utf-8") + print(json.dumps(receipt, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_leoclean_nosend_runtime.py b/tests/test_leoclean_nosend_runtime.py new file mode 100644 index 0000000..9fc6a74 --- /dev/null +++ b/tests/test_leoclean_nosend_runtime.py @@ -0,0 +1,757 @@ +"""Fail-closed contracts for the source-pinned leoclean no-send runtime.""" + +from __future__ import annotations + +import ast +import copy +import importlib.util +import io +import json +import re +import shutil +import subprocess +import sys +import tarfile +from pathlib import Path +from types import SimpleNamespace + +import pytest +import yaml + +from scripts import compile_leoclean_nosend_runtime as compiler +from scripts import leo_behavior_manifest +from scripts import verify_leoclean_nosend_runtime as verifier + +ROOT = Path(__file__).resolve().parents[1] +RUNTIME = ROOT / "hermes-agent" / "runtime" / "leoclean-nosend" +LOCK = json.loads((RUNTIME / "runtime-lock.json").read_text(encoding="utf-8")) +CONTRACT = json.loads((RUNTIME / "runtime-contract.json").read_text(encoding="utf-8")) + + +def load_bootstrap(): + spec = importlib.util.spec_from_file_location("leoclean_nosend_bootstrap_test", RUNTIME / "bootstrap.py") + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def load_module(path: Path, name: str): + spec = importlib.util.spec_from_file_location(name, path) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +def copy_profile(tmp_path: Path, bootstrap) -> Path: + runtime_root = tmp_path / "runtime" + runtime_root.mkdir(parents=True) + compiler._copy_profile(ROOT, runtime_root, LOCK) + bootstrap._runtime_root = lambda: runtime_root + profile = tmp_path / "profile" + shutil.copytree(runtime_root / "profile-template", profile) + return profile + + +def test_committed_lock_and_contract_are_exactly_pinned() -> None: + compiler._validate_lock(LOCK) + load_bootstrap()._validate_contract(CONTRACT) + assert LOCK["hermes"]["commit"] == "b2f477a30b3c05d0f383c543af98496ae8a96070" + assert LOCK["hermes"]["archive_sha256"] == "a1c3455e65d7948746046314c69bf39833d915ca6da75d74e47a89289a36c742" + assert LOCK["base_images"]["python"].startswith("python:3.11.9-slim-bookworm@sha256:") + assert CONTRACT["runtime_mode"] == "gcp_staging_no_send" + assert CONTRACT["transport_authority"] == "absent" + assert CONTRACT["toolset"]["allowed_tools"] == ["skills_list", "skill_view", "terminal"] + assert "send_message" in CONTRACT["toolset"]["forbidden_tools"] + assert CONTRACT["database"]["runtime_role"] == "leoclean_kb_runtime" + assert CONTRACT["database"]["automatic_context_modes"] == [ + "claim_evidence_challenge", + "review_only_candidate", + ] + assert CONTRACT["database"]["unsupported_context_policy"] == "fail_closed" + assert CONTRACT["profile"]["learning_policy"] == { + "automatic_memory_review_interval": 0, + "automatic_skill_review_interval": 0, + "memory_enabled": False, + "user_profile_enabled": False, + } + assert CONTRACT["profile"]["routing_policy"] == { + "cheap_model": "absent", + "smart_model_routing": False, + } + config = yaml.safe_load((RUNTIME / "profile-template" / "config.yaml").read_text(encoding="utf-8")) + assert config["memory"] == { + "enabled": False, + "memory_enabled": False, + "nudge_interval": 0, + "user_profile_enabled": False, + } + assert config["skills"] == {"creation_nudge_interval": 0} + assert config["smart_model_routing"] is False + assert leo_behavior_manifest.safe_model_config(RUNTIME / "profile-template" / "config.yaml")["status"] == ( + "observed" + ) + identity = json.loads((RUNTIME / "profile-template" / "gcp-runtime-identity.json").read_text(encoding="utf-8")) + assert identity == { + "expected_service_account": "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com", + "schema": "livingip.leocleanGcpRuntimeIdentity.v1", + } + + +def test_profile_sources_keep_gcp_adapters_out_of_vps_deploy_paths() -> None: + by_target = {entry["target"]: entry for entry in LOCK["profile_sources"]} + assert by_target["bin/cloudsql_memory_tool.py"]["source"] == ( + "hermes-agent/runtime/leoclean-nosend/profile-sources/cloudsql_memory_runtime.py" + ) + assert by_target["plugins/leo-db-context/__init__.py"]["source"] == ( + "hermes-agent/runtime/leoclean-nosend/profile-sources/leo_db_context.py" + ) + assert by_target["bin/cloudsql_memory_base.py"]["source"] == ("hermes-agent/leoclean-bin/cloudsql_memory_tool.py") + assert by_target["bin/operational_contracts.py"]["source"] == ( + "hermes-agent/runtime/leoclean-nosend/profile-sources/operational_contracts.py" + ) + assert by_target["plugins/leo-db-context/base.py"]["source"] == ( + "hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py" + ) + + +@pytest.mark.parametrize( + ("section", "field", "unsafe"), + [ + ("toolset", "allowed_tools", ["skills_list", "skill_view", "terminal", "send_message"]), + ("toolset", "forbidden_tools", ["process"]), + ("profile", "forbidden_entries", [".env"]), + ("environment", "forbidden_exact", ["TELEGRAM_BOT_TOKEN"]), + ("database", "direct_canonical_writes", "allowed"), + ], +) +def test_runtime_rejects_a_weakened_contract(section: str, field: str, unsafe: object) -> None: + bootstrap = load_bootstrap() + mutated = copy.deepcopy(CONTRACT) + mutated[section][field] = unsafe + + with pytest.raises(bootstrap.NoSendContractError): + bootstrap._validate_contract(mutated) + + +@pytest.mark.parametrize( + ("section", "index", "field", "unsafe"), + [ + ("patches", 0, "source", "/tmp/patch.py"), + ("patches", 0, "source", "../../patch.py"), + ("patches", 0, "target", "/tmp/run_agent.py"), + ("patches", 0, "target", "../../run_agent.py"), + ("profile_sources", 0, "source", "/etc/passwd"), + ("profile_sources", 0, "source", "../../outside"), + ("profile_sources", 0, "target", "/tmp/config.yaml"), + ("profile_sources", 0, "target", "../config.yaml"), + ("profile_sources", 0, "mode", "0777"), + ], +) +def test_lock_rejects_escaping_paths_and_unsafe_modes( + section: str, + index: int, + field: str, + unsafe: str, +) -> None: + mutated = copy.deepcopy(LOCK) + mutated[section][index][field] = unsafe + + with pytest.raises(compiler.CompileError): + compiler._validate_lock(mutated) + + +def test_lock_rejects_non_hex_hash_and_unbound_archive_url() -> None: + bad_hash = copy.deepcopy(LOCK) + bad_hash["hermes"]["archive_sha256"] = "z" * 64 + with pytest.raises(compiler.CompileError, match="archive_sha256"): + compiler._validate_lock(bad_hash) + + bad_url = copy.deepcopy(LOCK) + bad_url["hermes"]["archive_url"] = "https://example.invalid/archive.tar.gz" + with pytest.raises(compiler.CompileError, match="archive URL"): + compiler._validate_lock(bad_url) + + +@pytest.mark.parametrize("kind", ["traversal", "absolute", "symlink", "hardlink", "device"]) +def test_archive_extraction_rejects_unsafe_members(tmp_path: Path, kind: str) -> None: + archive = tmp_path / "unsafe.tar.gz" + root = LOCK["hermes"]["archive_root"] + with tarfile.open(archive, "w:gz") as bundle: + if kind == "traversal": + member = tarfile.TarInfo(f"{root}/../../escape") + elif kind == "absolute": + member = tarfile.TarInfo("/tmp/escape") + elif kind == "symlink": + member = tarfile.TarInfo(f"{root}/link") + member.type = tarfile.SYMTYPE + member.linkname = "/etc/passwd" + elif kind == "hardlink": + member = tarfile.TarInfo(f"{root}/hardlink") + member.type = tarfile.LNKTYPE + member.linkname = f"{root}/target" + else: + member = tarfile.TarInfo(f"{root}/device") + member.type = tarfile.CHRTYPE + if member.isreg(): + payload = b"unsafe" + member.size = len(payload) + bundle.addfile(member, io.BytesIO(payload)) + else: + bundle.addfile(member) + + with pytest.raises(compiler.CompileError): + compiler._safe_extract(archive, tmp_path / "output", root) + + +def test_profile_template_and_runtime_state_are_exact(tmp_path: Path) -> None: + bootstrap = load_bootstrap() + profile = copy_profile(tmp_path, bootstrap) + bootstrap._validate_profile(profile, CONTRACT, template=True) + + for name in CONTRACT["profile"]["required_runtime_entries"]: + (profile / name).write_text("{}\n", encoding="utf-8") + for name in ("logs", "sessions", "platforms"): + (profile / name).mkdir() + (profile / "state.db").write_bytes(b"sqlite fixture") + bootstrap._validate_profile(profile, CONTRACT, template=False) + + +@pytest.mark.parametrize( + ("section", "field", "unsafe"), + [ + ("memory", "memory_enabled", True), + ("memory", "memory_enabled", 0), + ("memory", "user_profile_enabled", True), + ("memory", "nudge_interval", 10), + ("memory", "nudge_interval", False), + ("skills", "creation_nudge_interval", 10), + ("skills", "creation_nudge_interval", False), + ], +) +def test_profile_rejects_learning_policy_drift( + tmp_path: Path, + section: str, + field: str, + unsafe: object, +) -> None: + bootstrap = load_bootstrap() + profile = copy_profile(tmp_path, bootstrap) + config_path = profile / "config.yaml" + config = yaml.safe_load(config_path.read_text(encoding="utf-8")) + config[section][field] = unsafe + config_path.write_text(yaml.safe_dump(config, sort_keys=False), encoding="utf-8") + + with pytest.raises(bootstrap.NoSendContractError): + bootstrap._load_profile_config(profile, CONTRACT) + with pytest.raises(verifier.VerificationError): + verifier._validate_profile_config(profile, CONTRACT) + + +def test_profile_rejects_smart_model_routing(tmp_path: Path) -> None: + bootstrap = load_bootstrap() + profile = copy_profile(tmp_path, bootstrap) + config_path = profile / "config.yaml" + config = yaml.safe_load(config_path.read_text(encoding="utf-8")) + config["smart_model_routing"] = { + "enabled": True, + "cheap_model": {"provider": "openrouter", "model": "unreviewed/model"}, + } + config_path.write_text(yaml.safe_dump(config, sort_keys=False), encoding="utf-8") + + with pytest.raises(bootstrap.NoSendContractError, match="smart-model routing"): + bootstrap._load_profile_config(profile, CONTRACT) + with pytest.raises(verifier.VerificationError, match="smart-model routing"): + verifier._validate_profile_config(profile, CONTRACT) + + +def test_real_cloudsql_producer_feeds_gcp_plugin_with_private_bounded_excerpts(tmp_path: Path) -> None: + bootstrap = load_bootstrap() + profile = copy_profile(tmp_path, bootstrap) + cloudsql = load_module(profile / "bin" / "cloudsql_memory_tool.py", "leoclean_cloudsql_runtime_test") + plugin = load_module(profile / "plugins" / "leo-db-context" / "__init__.py", "leoclean_db_plugin_test") + query = "Challenge one exact claim body with its evidence source and offer a narrower revision." + result = { + "artifact": "teleo_cloudsql_canonical_kb_context", + "backend": "cloudsql:teleo-pgvector-standby/teleo_canonical/public+kb_stage", + "query": query, + "terms": ["challenge", "claim", "evidence"], + "claims": [ + { + "id": "11111111-1111-1111-1111-111111111111", + "type": "fact", + "text": "Exact canonical claim body.", + "status": "open", + "confidence": 0.6, + "tags": ["test"], + "score": 3, + "evidence": [ + { + "source_id": "22222222-2222-2222-2222-222222222222", + "role": "supports", + "source_type": "document", + "excerpt": "Exact linked evidence excerpt.", + } + ], + "edges": [], + } + ], + "context_rows": [], + "hit_count_total": 1, + "hits": [], + "privacy": "private agent context mode", + } + args = SimpleNamespace(command="context", canonical_db="teleo_canonical") + payload = cloudsql._enrich_context(args, query, result) + contract_ids = {item["id"] for item in payload["operational_contracts"]} + assert "claim_evidence_challenge" in contract_ids + assert 'Exact claim body: "Exact canonical claim body."' in payload["compiled_response"] + assert 'Exact evidence excerpt: "Exact linked evidence excerpt."' in payload["compiled_response"] + + observed = {} + + def fake_runner(command, **kwargs): + observed.update({"command": command, **kwargs}) + return subprocess.CompletedProcess(command, 0, stdout=json.dumps(payload), stderr="") + + snapshot = plugin._load_database_snapshot(query, hermes_home=profile, runner=fake_runner) + assert snapshot["status"] == "ok" + assert snapshot["requires_database_truth"] is True + assert "Exact canonical claim body." in snapshot["context"] + assert "Exact linked evidence excerpt." in snapshot["context"] + assert observed["command"][:3] == [str(profile / "bin" / "teleo-kb"), "context", query] + assert "--include-excerpts" in observed["command"] + assert "--local" not in observed["command"] + assert observed["check"] is False + + +def test_gcp_runtime_identity_parser_accepts_only_the_exact_staging_account(tmp_path: Path) -> None: + bootstrap = load_bootstrap() + profile = copy_profile(tmp_path, bootstrap) + cloudsql = load_module(profile / "bin" / "cloudsql_memory_tool.py", "leoclean_identity_parser_test") + valid = ( + b'{"expected_service_account":"sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com",' + b'"schema":"livingip.leocleanGcpRuntimeIdentity.v1"}' + ) + assert cloudsql._parse_runtime_identity(valid) == cloudsql.STAGING_SERVICE_ACCOUNT + for invalid in ( + b'{"expected_service_account":"unknown@teleo-501523.iam.gserviceaccount.com","schema":"livingip.leocleanGcpRuntimeIdentity.v1"}', + b'{"expected_service_account":"sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com","schema":"wrong"}', + b'{"expected_service_account":"sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com","schema":"livingip.leocleanGcpRuntimeIdentity.v1","extra":true}', + b'{"expected_service_account":"one","expected_service_account":"two","schema":"livingip.leocleanGcpRuntimeIdentity.v1"}', + ): + with pytest.raises(SystemExit, match=r"Runtime identity manifest .*invalid"): + cloudsql._parse_runtime_identity(invalid) + + +def test_packaged_current_main_contract_keeps_review_only_candidate_out_of_database_intents( + tmp_path: Path, +) -> None: + bootstrap = load_bootstrap() + profile = copy_profile(tmp_path, bootstrap) + cloudsql = load_module(profile / "bin" / "cloudsql_memory_tool.py", "leoclean_candidate_contract_test") + ids = { + item["id"] + for item in cloudsql.contracts.operational_contracts( + "Turn the surviving claim into a review-only proposal: exact body, confidence, falsifier, required " + "sources, and links to claims that support or contradict it. Do not stage or write anything." + ) + } + assert ids == {"reply_budget"} + + +def test_sql_free_contract_compiler_matches_current_main_for_supported_modes(tmp_path: Path) -> None: + bootstrap = load_bootstrap() + profile = copy_profile(tmp_path, bootstrap) + cloudsql = load_module(profile / "bin" / "cloudsql_memory_tool.py", "leoclean_contract_parity_test") + current_main = load_module(ROOT / "hermes-agent" / "leoclean-bin" / "kb_tool.py", "kb_tool_contract_parity_test") + prompts = [ + "Summarize this in under 80 words.", + ( + "Turn the surviving claim into a review-only proposal: exact body, confidence, falsifier, required " + "sources, and links to claims that support or contradict it. Do not stage or write anything." + ), + "Challenge one exact claim body with its evidence source and offer a narrower revision.", + ] + for prompt in prompts: + assert cloudsql.contracts.operational_contracts(prompt) == current_main.operational_contracts(prompt) + + source_path = profile / "bin" / "operational_contracts.py" + source = source_path.read_text(encoding="utf-8") + ast.parse(source) + forbidden = re.compile(r"\b(?:delete|insert|psql|select|socket|subprocess|truncate|update)\b", re.I) + assert not forbidden.search(source) + + +def test_gcp_plugin_preserves_current_main_subject_anchors(tmp_path: Path) -> None: + bootstrap = load_bootstrap() + profile = copy_profile(tmp_path, bootstrap) + plugin = load_module(profile / "plugins" / "leo-db-context" / "__init__.py", "leoclean_subject_plugin_test") + query = "How should we revise that conclusion?" + history = [ + {"role": "user", "content": "Compare LivingIP ownership models."}, + {"role": "assistant", "content": "I found a claim to inspect."}, + {"role": "user", "content": query}, + ] + observed = {} + + def fake_runner(command, **kwargs): + observed.update({"command": command, **kwargs}) + payload = { + "claims": [], + "compiled_response": None, + "context_rows": [], + "operational_contracts": [{"id": "reply_budget", "hard_max_words": 220}], + "runtime_contract_mode": "general", + } + return subprocess.CompletedProcess(command, 0, stdout=json.dumps(payload), stderr="") + + snapshot = plugin._load_database_snapshot( + query, + conversation_history=history, + hermes_home=profile, + runner=fake_runner, + ) + command = observed["command"] + assert snapshot["status"] == "ok" + assert command[command.index("--retrieval-query") + 1] == query + anchors = [command[index + 1] for index, item in enumerate(command) if item == "--retrieval-anchor"] + assert "LivingIP" in anchors + assert "--include-excerpts" in command + + +def test_gcp_plugin_fails_closed_for_unimplemented_database_lifecycle_contract(tmp_path: Path) -> None: + bootstrap = load_bootstrap() + profile = copy_profile(tmp_path, bootstrap) + plugin = load_module(profile / "plugins" / "leo-db-context" / "__init__.py", "leoclean_gap_plugin_test") + query = "Which database proposal is currently approved or applied?" + + def fake_runner(command, **_kwargs): + payload = { + "claims": [], + "compiled_response": None, + "context_rows": [], + "operational_contracts": [{"id": "reply_budget", "hard_max_words": 220}], + "runtime_contract_mode": "general", + } + return subprocess.CompletedProcess(command, 0, stdout=json.dumps(payload), stderr="") + + snapshot = plugin._load_database_snapshot(query, hermes_home=profile, runner=fake_runner) + assert snapshot["status"] == "error" + assert snapshot["requires_database_truth"] is True + assert "read-only current database contract lookup failed" in snapshot["context"] + + +def test_gcp_plugin_registers_callbacks_with_the_gcp_loader_installed(tmp_path: Path) -> None: + bootstrap = load_bootstrap() + profile = copy_profile(tmp_path, bootstrap) + plugin = load_module(profile / "plugins" / "leo-db-context" / "__init__.py", "leoclean_registered_plugin_test") + registered = {} + + class Context: + def register_hook(self, name, callback): + registered[name] = callback + + plugin.register(Context()) + assert plugin.base._load_database_snapshot is plugin._load_database_snapshot + assert set(registered) == {"post_llm_call", "pre_llm_call"} + + +def test_gcp_cloudsql_runtime_backfills_subject_claim_without_losing_raw_results(tmp_path: Path) -> None: + bootstrap = load_bootstrap() + profile = copy_profile(tmp_path, bootstrap) + cloudsql = load_module(profile / "bin" / "cloudsql_memory_tool.py", "leoclean_anchor_backfill_test") + query = "How should we revise that conclusion?" + raw_claim = {"id": "raw", "text": "A cross-domain claim", "tags": [], "evidence": [], "edges": []} + subject_claim = { + "id": "subject", + "text": "LivingIP ownership claim", + "tags": ["LivingIP"], + "evidence": [], + "edges": [], + } + calls = [] + + def fake_query(_args, semantic_query, _limit): + calls.append(semantic_query) + claims = [subject_claim] if semantic_query == "LivingIP" else [raw_claim] + return { + "artifact": "teleo_cloudsql_canonical_kb_context", + "claims": claims, + "context_rows": [], + "hit_count_total": len(claims), + "hits": [ + {"source_table": "public.claims", "row_id": claim["id"], "excerpt": claim["text"]} for claim in claims + ], + "privacy": "private agent context mode: short excerpts included", + "query": semantic_query, + } + + cloudsql._BASE_QUERY_ROWS = fake_query + args = SimpleNamespace( + command="context", + canonical_db="teleo_canonical", + retrieval_anchor=["LivingIP"], + retrieval_query=query, + ) + payload = cloudsql._query_rows(args, query, 4) + assert calls == [query, "LivingIP"] + assert [claim["id"] for claim in payload["claims"]] == ["subject", "raw"] + assert payload["query"] == query + assert payload["retrieval_query_sha256"] + assert payload["retrieval_anchors_sha256"] + + +def test_gcp_cloudsql_parser_accepts_only_bounded_contextual_retrieval_flags( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + bootstrap = load_bootstrap() + profile = copy_profile(tmp_path, bootstrap) + cloudsql = load_module(profile / "bin" / "cloudsql_memory_tool.py", "leoclean_context_parser_test") + monkeypatch.setattr( + cloudsql.sys, + "argv", + [ + "cloudsql_memory_tool.py", + "context", + "revise that", + "--retrieval-query", + "revise that", + "--retrieval-anchor", + "LivingIP", + "--include-excerpts", + ], + ) + parsed = cloudsql._parse_args() + assert parsed.query == "revise that" + assert parsed.retrieval_query == "revise that" + assert parsed.retrieval_anchor == ["LivingIP"] + assert parsed.include_excerpts is True + + monkeypatch.setattr( + cloudsql.sys, + "argv", + [ + "cloudsql_memory_tool.py", + "context", + "revise that", + "--retrieval-query", + "first", + "--retrieval-query", + "second", + ], + ) + with pytest.raises(SystemExit, match="Only one retrieval query"): + cloudsql._parse_args() + + +@pytest.mark.parametrize("entry", [".env", "BOOT.md", "gateway.json", "hooks", "processes.json", "secrets"]) +def test_runtime_profile_rejects_startup_control_entries(tmp_path: Path, entry: str) -> None: + bootstrap = load_bootstrap() + profile = copy_profile(tmp_path, bootstrap) + for name in CONTRACT["profile"]["required_runtime_entries"]: + (profile / name).write_text("{}\n", encoding="utf-8") + target = profile / entry + if "." in entry and entry != "hooks": + target.write_text("unsafe\n", encoding="utf-8") + else: + target.mkdir() + + with pytest.raises(bootstrap.NoSendContractError): + bootstrap._validate_profile(profile, CONTRACT, template=False) + + +def test_profile_rejects_nested_mutation_and_symlink(tmp_path: Path) -> None: + bootstrap = load_bootstrap() + mutated = copy_profile(tmp_path / "mutated", bootstrap) + skill = mutated / "skills" / "teleo-kb-bridge" / "SKILL.md" + skill.write_text(skill.read_text(encoding="utf-8") + "\nunsafe\n", encoding="utf-8") + with pytest.raises(bootstrap.NoSendContractError, match="immutable profile tree drifted"): + bootstrap._validate_profile(mutated, CONTRACT, template=True) + + linked = copy_profile(tmp_path / "linked", bootstrap) + (linked / "skills" / "escape").symlink_to("/tmp") + with pytest.raises(bootstrap.NoSendContractError, match="symlink"): + bootstrap._validate_profile(linked, CONTRACT, template=True) + + +def forbidden_environment_cases() -> list[str]: + policy = CONTRACT["environment"] + cases = list(policy["forbidden_exact"]) + cases.extend(f"{prefix}INJECTED" for prefix in policy["forbidden_prefixes"]) + cases.extend(f"INJECTED{suffix}" for suffix in policy["forbidden_suffixes"]) + return sorted(set(cases)) + + +@pytest.mark.parametrize("name", forbidden_environment_cases()) +def test_each_forbidden_environment_shape_fails_closed(monkeypatch: pytest.MonkeyPatch, name: str) -> None: + bootstrap = load_bootstrap() + monkeypatch.setattr(bootstrap.os, "environ", {name: "secret-shaped-sentinel"}) + + with pytest.raises(bootstrap.NoSendContractError) as caught: + bootstrap._validate_environment(CONTRACT) + + assert name in str(caught.value) + assert "secret-shaped-sentinel" not in str(caught.value) + + +def test_only_reviewed_provider_and_runtime_environment_are_accepted(monkeypatch: pytest.MonkeyPatch) -> None: + bootstrap = load_bootstrap() + monkeypatch.setattr( + bootstrap.os, + "environ", + { + "OPENROUTER_API_KEY": "secret-shaped-sentinel", + "HERMES_HOME": "/profile", + "PYTHONNOUSERSITE": "1", + }, + ) + bootstrap._validate_environment(CONTRACT) + + +def test_sealed_tool_map_rejects_registration_and_overwrite() -> None: + bootstrap = load_bootstrap() + tools = bootstrap._SealedToolMap({"terminal": object()}) + + with pytest.raises(bootstrap.NoSendContractError): + tools["send_message"] = object() + with pytest.raises(bootstrap.NoSendContractError): + tools["terminal"] = object() + with pytest.raises(bootstrap.NoSendContractError): + tools.update({"process": object()}) + assert list(tools) == ["terminal"] + + +def test_restricted_terminal_uses_exact_argv_and_sanitized_environment( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + bootstrap = load_bootstrap() + profile = tmp_path / "profile" + wrapper = profile / "bin" / "teleo-kb" + wrapper.parent.mkdir(parents=True) + wrapper.write_text("#!/bin/sh\n", encoding="utf-8") + wrapper.chmod(0o755) + observed = {} + + def fake_run(argv, **kwargs): + observed.update({"argv": argv, **kwargs}) + return subprocess.CompletedProcess(argv, 0, stdout="ok", stderr="") + + monkeypatch.setattr(bootstrap.subprocess, "run", fake_run) + monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "must-not-propagate") + monkeypatch.setenv("PGPASSWORD", "must-not-propagate") + result = json.loads( + bootstrap._restricted_terminal_handler( + profile, + set(CONTRACT["toolset"]["terminal_commands"]), + {"command": "teleo-kb search 'claim; literal' --limit 2", "timeout": 500}, + ) + ) + + assert observed["argv"] == [str(wrapper.resolve()), "search", "claim; literal", "--limit", "2"] + assert observed["cwd"] == profile + assert observed["timeout"] == 120 + assert observed["env"] == { + "HOME": str(profile / "state"), + "HERMES_HOME": str(profile), + "LANG": "C", + "LC_ALL": "C", + "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", + } + assert result == {"error": None, "exit_code": 0, "output": "ok", "truncated": False} + + +@pytest.mark.parametrize( + "tool_args", + [ + {"command": "sh -c whoami"}, + {"command": "teleo-kb apply-proposal 123"}, + {"command": "teleo-kb status ; whoami"}, + {"command": "teleo-kb status --user=postgres"}, + {"command": "teleo-kb status\nwhoami"}, + {"command": "teleo-kb status", "workdir": "/tmp"}, + {"command": "teleo-kb status", "timeout": "not-an-integer"}, + ], +) +def test_restricted_terminal_rejects_bypasses_without_subprocess( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + tool_args: dict[str, object], +) -> None: + bootstrap = load_bootstrap() + profile = tmp_path / "profile" + wrapper = profile / "bin" / "teleo-kb" + wrapper.parent.mkdir(parents=True) + wrapper.write_text("#!/bin/sh\n", encoding="utf-8") + wrapper.chmod(0o755) + + def forbidden_call(*_args, **_kwargs): + pytest.fail("subprocess must not be called for a rejected terminal request") + + monkeypatch.setattr(bootstrap.subprocess, "run", forbidden_call) + result = json.loads( + bootstrap._restricted_terminal_handler( + profile, + set(CONTRACT["toolset"]["terminal_commands"]), + tool_args, + ) + ) + assert result["exit_code"] == 126 + + +def test_verifier_preserves_virtual_environment_python_symlink( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + interpreter = tmp_path / "venv" / "bin" / "python" + interpreter.parent.mkdir(parents=True) + interpreter.symlink_to(sys.executable) + observed = {} + + def fake_verify_artifact(**kwargs): + observed.update(kwargs) + return {"schema": verifier.RECEIPT_SCHEMA, "status": "pass"} + + monkeypatch.setattr(verifier, "verify_artifact", fake_verify_artifact) + monkeypatch.setattr( + sys, + "argv", + [ + "verify", + "--repo-root", + str(ROOT), + "--artifact", + str(tmp_path / "artifact"), + "--runtime-python", + str(interpreter), + ], + ) + + assert verifier.main() == 0 + assert observed["runtime_python"] == interpreter.absolute() + assert observed["runtime_python"] != interpreter.resolve() + assert json.loads(capsys.readouterr().out)["status"] == "pass" + + +def test_ci_runs_the_release_canary_and_retains_its_receipt() -> None: + workflow = (ROOT / ".github" / "workflows" / "ci.yml").read_text(encoding="utf-8") + wrapper = (ROOT / "scripts" / "run_leoclean_nosend_runtime_canary.sh").read_text(encoding="utf-8") + + assert "leoclean-nosend-runtime:" in workflow + assert "scripts/run_leoclean_nosend_runtime_canary.sh" in workflow + assert 'python-version: "3.11.9"' in workflow + assert "uv==0.9.30" in workflow + assert ".crabbox-results/leoclean-nosend-runtime.json" in workflow + assert "--development" not in wrapper + assert "--allow-dirty" not in wrapper + assert 'rm -f "$receipt"' in wrapper + assert 'repeat_artifact="$temporary/artifact-repeat"' in wrapper + assert "first != second" in wrapper + assert "compile-primary.json" in wrapper + assert '--runtime-python "$runtime_venv/bin/python"' in wrapper + assert 'receipt.get("verification_mode") != "release"' in wrapper