diff --git a/scripts/leo_oos_readonly_guard.py b/scripts/leo_oos_readonly_guard.py index a11f617..23647f4 100644 --- a/scripts/leo_oos_readonly_guard.py +++ b/scripts/leo_oos_readonly_guard.py @@ -3,158 +3,251 @@ from __future__ import annotations -import argparse +import hashlib import json import os -import shlex -import shutil +import signal import subprocess +import sys import uuid from pathlib import Path from typing import Any -READ_ONLY_BRIDGE_COMMANDS = frozenset( - { - "search", - "context", - "show", - "evidence", - "edges", - "list-proposals", - "search-proposals", - "show-proposal", - "status", - "decision-matrix-status", +STRUCTURED_READ_TOOL_NAME = "teleo-kb" +BRIDGE_TIMEOUT_SECONDS = 45 +BRIDGE_TERMINATE_GRACE_SECONDS = 5 +PROPOSAL_STATUSES = ("pending_review", "approved", "applied", "canceled", "rejected", "all") + +_FIELD_SCHEMAS = { + "query": { + "type": "string", + "minLength": 1, + "maxLength": 4096, + "description": "Natural-language lookup text.", + }, + "claim_id": { + "type": "string", + "description": "Canonical claim UUID.", + }, + "proposal_id": { + "type": "string", + "description": "Proposal UUID.", + }, + "status": { + "type": "string", + "enum": list(PROPOSAL_STATUSES), + "description": "Proposal status filter.", + }, + "limit": { + "type": "integer", + "minimum": 1, + "maximum": 100, + "description": "Maximum rows to return.", + }, + "context_limit": { + "type": "integer", + "minimum": 1, + "maximum": 100, + "description": "Maximum identity/context rows.", + }, + "format": { + "type": "string", + "enum": ["markdown", "json"], + "description": "Response format.", + }, +} +_ACTION_SPECS = { + "search": {"required": ("query",), "optional": ("limit", "context_limit", "format")}, + "context": {"required": ("query",), "optional": ("limit", "context_limit", "format")}, + "show": {"required": ("claim_id",), "optional": ("format",)}, + "evidence": {"required": ("claim_id",), "optional": ("limit", "format")}, + "edges": {"required": ("claim_id",), "optional": ("limit", "format")}, + "list-proposals": {"required": (), "optional": ("status", "limit", "format")}, + "search-proposals": {"required": ("query",), "optional": ("status", "limit", "format")}, + "show-proposal": {"required": ("proposal_id",), "optional": ("format",)}, + "status": {"required": (), "optional": ("format",)}, + "decision-matrix-status": {"required": (), "optional": ("format",)}, +} +READ_ONLY_BRIDGE_COMMANDS = frozenset(_ACTION_SPECS) + + +def _action_schema(action: str) -> dict[str, Any]: + spec = _ACTION_SPECS[action] + fields = (*spec["required"], *spec["optional"]) + return { + "type": "object", + "properties": { + "action": {"type": "string", "enum": [action]}, + **{field: _FIELD_SCHEMAS[field] for field in fields}, + }, + "required": ["action", *spec["required"]], + "additionalProperties": False, } -) -HARMLESS_TRAILING_REDIRECT_TOKENS = frozenset({"2>/dev/null", "2> /dev/null"}) + + +STRUCTURED_READ_TOOL_SCHEMA = { + "name": STRUCTURED_READ_TOOL_NAME, + "description": ( + "Read Leo's canonical Teleo knowledge base through a bounded, read-only interface. " + "This tool cannot stage, approve, apply, or mutate knowledge." + ), + "parameters": { + "type": "object", + "oneOf": [_action_schema(action) for action in sorted(_ACTION_SPECS)], + }, +} class ReadOnlyGuardError(RuntimeError): """The requested command cannot be proven read-only.""" -class GuardArgumentParser(argparse.ArgumentParser): - def error(self, message: str) -> None: - raise ReadOnlyGuardError(message) - - def exit(self, status: int = 0, message: str | None = None) -> None: - raise ReadOnlyGuardError(message or f"argument parser exited with status {status}") - - def _uuid_argument(value: str) -> str: try: return str(uuid.UUID(value)) except ValueError as exc: - raise argparse.ArgumentTypeError("expected a UUID") from exc + raise ReadOnlyGuardError("expected a UUID") from exc -def validate_read_only_bridge_argv(argv: list[str]) -> None: - """Accept only the bridge's bounded read commands and exact arguments.""" - - parser = GuardArgumentParser(add_help=False) - subparsers = parser.add_subparsers(dest="command", required=True) - - def command(name: str) -> argparse.ArgumentParser: - item = subparsers.add_parser(name, add_help=False) - item.add_argument("--help", action="store_true") - item.add_argument("--format", choices=("markdown", "json"), default="markdown") - return item - - for name in ("search", "context"): - item = command(name) - item.add_argument("query", nargs="+") - item.add_argument("--limit", type=int, default=5) - item.add_argument("--context-limit", type=int, default=5 if name == "search" else 6) - show = command("show") - show.add_argument("claim_id", type=_uuid_argument) - for name in ("evidence", "edges"): - item = command(name) - item.add_argument("claim_id", type=_uuid_argument) - item.add_argument("--limit", type=int, default=12) - list_proposals = command("list-proposals") - list_proposals.add_argument( - "--status", - choices=("pending_review", "approved", "applied", "canceled", "rejected", "all"), - default="pending_review", - ) - list_proposals.add_argument("--limit", type=int, default=10) - search_proposals = command("search-proposals") - search_proposals.add_argument("query", nargs="+") - search_proposals.add_argument( - "--status", - choices=("pending_review", "approved", "applied", "canceled", "rejected", "all"), - default="all", - ) - search_proposals.add_argument("--limit", type=int, default=10) - show_proposal = command("show-proposal") - show_proposal.add_argument("proposal_id", type=_uuid_argument) - command("status") - command("decision-matrix-status") - parsed = parser.parse_args(argv) - for name, value in vars(parsed).items(): - if name.endswith("limit") and not 1 <= value <= 100: - raise ReadOnlyGuardError(f"{name.replace('_', '-')} must be between 1 and 100") +def _required_text(tool_args: dict[str, Any], field: str, *, max_length: int = 4096) -> str: + value = tool_args.get(field) + if not isinstance(value, str) or not value.strip(): + raise ReadOnlyGuardError(f"{field} is required") + normalized = value.strip() + if len(normalized) > max_length: + raise ReadOnlyGuardError(f"{field} is too long") + return normalized -def classify_terminal_command( - wrapper: Path, - temp_profile: Path, - tool_args: dict[str, Any], - *, - allow_kb_reads: bool, -) -> tuple[list[str] | None, str | None]: - """Return verified argv or a fail-closed reason without executing anything.""" +def _bounded_integer(tool_args: dict[str, Any], field: str) -> int: + value = tool_args.get(field) + if not isinstance(value, int) or isinstance(value, bool) or not 1 <= value <= 100: + raise ReadOnlyGuardError(f"{field} must be an integer between 1 and 100") + return value - command = str(tool_args.get("command") or "") - if len(command) > 8192: - return None, "command is too long" - if tool_args.get("background") or tool_args.get("pty") or tool_args.get("workdir"): - return None, "background, PTY, and workdir overrides are disabled" - if not allow_kb_reads: - return None, "database tools are disabled for the no-DB ablation" + +def _enum_text(tool_args: dict[str, Any], field: str, choices: tuple[str, ...]) -> str: + value = tool_args.get(field) + if not isinstance(value, str) or value not in choices: + raise ReadOnlyGuardError(f"{field} is not an allowed value") + return value + + +def structured_read_argv(tool_args: dict[str, Any]) -> list[str]: + """Compile typed arguments into bridge argv without an executable.""" + + if not isinstance(tool_args, dict): + raise ReadOnlyGuardError("tool arguments must be an object") + action = tool_args.get("action") + if not isinstance(action, str) or action not in _ACTION_SPECS: + raise ReadOnlyGuardError("action must name an available read-only KB operation") + spec = _ACTION_SPECS[action] + allowed_fields = {"action", *spec["required"], *spec["optional"]} + unexpected = sorted(set(tool_args) - allowed_fields) + if unexpected: + raise ReadOnlyGuardError(f"fields are not valid for {action}: {', '.join(unexpected)}") + for field in spec["required"]: + _required_text(tool_args, field, max_length=64 if field.endswith("_id") else 4096) + + argv = [action] + if action in {"search", "context"}: + argv.append(_required_text(tool_args, "query")) + elif action in {"show", "evidence", "edges"}: + argv.append(_uuid_argument(_required_text(tool_args, "claim_id", max_length=64))) + elif action == "search-proposals": + argv.append(_required_text(tool_args, "query")) + elif action == "show-proposal": + argv.append(_uuid_argument(_required_text(tool_args, "proposal_id", max_length=64))) + + if "status" in tool_args: + argv.extend(["--status", _enum_text(tool_args, "status", PROPOSAL_STATUSES)]) + if "limit" in tool_args: + argv.extend(["--limit", str(_bounded_integer(tool_args, "limit"))]) + if "context_limit" in tool_args: + argv.extend(["--context-limit", str(_bounded_integer(tool_args, "context_limit"))]) + if "format" in tool_args: + argv.extend(["--format", _enum_text(tool_args, "format", ("markdown", "json"))]) + return argv + + +def _canonical_sha256(value: Any) -> str: + encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _verified_kb_tool(temp_profile: Path, expected_sha256: str) -> Path: + candidate = temp_profile / "bin" / "kb_tool.py" + if ( + temp_profile.is_symlink() + or (temp_profile / "bin").is_symlink() + or candidate.is_symlink() + or not candidate.is_file() + ): + raise ReadOnlyGuardError("temporary-profile KB tool must be a regular non-symlink file") + resolved = candidate.resolve(strict=True) + if resolved.parent != (temp_profile / "bin").resolve(strict=True): + raise ReadOnlyGuardError("temporary-profile KB tool escaped the profile bin directory") + observed_sha256 = hashlib.sha256(resolved.read_bytes()).hexdigest() + if observed_sha256 != expected_sha256: + raise ReadOnlyGuardError("temporary-profile KB tool hash does not match the frozen source") + return resolved + + +def _terminate_process_group(proc: subprocess.Popen[str]) -> None: try: - argv = shlex.split(command) - except ValueError as exc: - return None, f"invalid command quoting: {exc}" - while argv and argv[-1] in HARMLESS_TRAILING_REDIRECT_TOKENS: - argv.pop() - command_path = f"{temp_profile / 'bin'}:{os.environ.get('PATH', '')}" - try: - exact_wrapper = wrapper.resolve(strict=True) - executable = shutil.which(argv[0], path=command_path) if argv else None - invoked = Path(executable).resolve(strict=True) if executable else None + os.killpg(proc.pid, signal.SIGTERM) except OSError: - invoked = None - exact_wrapper = wrapper.resolve(strict=False) - if invoked != exact_wrapper: - return None, "only the exact temporary-profile teleo-kb wrapper is available" - if len(argv) < 2 or argv[1] not in READ_ONLY_BRIDGE_COMMANDS: - return None, "only read-only Teleo KB bridge commands are available" + pass try: - validate_read_only_bridge_argv(argv[1:]) - except ReadOnlyGuardError as exc: - return None, f"invalid read-only KB arguments: {exc}" - return argv, None + proc.communicate(timeout=BRIDGE_TERMINATE_GRACE_SECONDS) + return + except subprocess.TimeoutExpired: + pass + try: + os.killpg(proc.pid, signal.SIGKILL) + except OSError: + pass + try: + proc.communicate(timeout=BRIDGE_TERMINATE_GRACE_SECONDS) + except (OSError, subprocess.TimeoutExpired): + pass -def restricted_bridge_terminal_handler( - wrapper: Path, +def restricted_structured_read_handler( + kb_tool: Path, temp_profile: Path, tool_args: dict[str, Any], *, allow_kb_reads: bool, + expected_kb_tool_sha256: str, **_kwargs: Any, ) -> str: - argv, reason = classify_terminal_command( - wrapper, - temp_profile, - tool_args, - allow_kb_reads=allow_kb_reads, - ) - if argv is None: - return json.dumps({"output": "", "exit_code": 126, "error": reason}) - timeout = min(max(int(tool_args.get("timeout") or 60), 1), 120) + """Execute one typed read without exposing a shell command surface.""" + + if not allow_kb_reads: + return json.dumps( + { + "output": "", + "exit_code": 126, + "error": "database tools are disabled for the no-DB ablation", + "error_category": "kb_reads_disabled", + } + ) + try: + exact_kb_tool = _verified_kb_tool(temp_profile, expected_kb_tool_sha256) + if kb_tool != exact_kb_tool: + raise ReadOnlyGuardError("configured KB tool is not the frozen temporary-profile tool") + argv = [str(Path(sys.executable).resolve(strict=True)), str(exact_kb_tool), "--local"] + argv.extend(structured_read_argv(tool_args)) + except (OSError, ReadOnlyGuardError) as exc: + return json.dumps( + { + "output": "", + "exit_code": 126, + "error": str(exc), + "error_category": "invalid_read_arguments", + } + ) child_environment = { "HOME": str(temp_profile), "HERMES_HOME": str(temp_profile), @@ -162,65 +255,104 @@ def restricted_bridge_terminal_handler( "TELEO_KB_MODE": "local", } try: - proc = subprocess.run( + proc = subprocess.Popen( argv, cwd=temp_profile, env=child_environment, text=True, - capture_output=True, - check=False, - timeout=timeout, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + start_new_session=True, ) + except OSError as exc: + return json.dumps( + { + "output": "", + "exit_code": 126, + "error": f"bridge launch failed: {type(exc).__name__}", + "error_category": "bridge_launch", + } + ) + try: + stdout, stderr = proc.communicate(timeout=BRIDGE_TIMEOUT_SECONDS) except subprocess.TimeoutExpired: - return json.dumps({"output": "", "exit_code": 124, "error": f"bridge command exceeded {timeout}s"}) + _terminate_process_group(proc) + return json.dumps( + { + "output": "", + "exit_code": 124, + "error": f"bridge command exceeded {BRIDGE_TIMEOUT_SECONDS}s", + "error_category": "bridge_timeout", + } + ) return json.dumps( { - "output": proc.stdout, + "output": stdout, "exit_code": proc.returncode, - "error": proc.stderr or None, + "error": stderr or None, + "error_category": None if proc.returncode == 0 and not stderr else "bridge_exit", } ) -def install_read_only_tool_surface(temp_profile: Path, *, allow_kb_reads: bool) -> dict[str, Any]: - """Replace Hermes' registry with skills plus one guarded terminal tool.""" +def install_read_only_tool_surface( + temp_profile: Path, + *, + allow_kb_reads: bool, + expected_kb_tool_sha256: str, +) -> dict[str, Any]: + """Replace Hermes' registry with skills plus one structured KB read tool.""" - import tools.skills_tool - import tools.terminal_tool # noqa: F401 + import tools.skills_tool # noqa: F401 - registers the two skill readers import toolsets from hermes_cli import tools_config from tools.registry import registry toolset_name = "leo-oos-kb-readonly" if allow_kb_reads else "leo-oos-no-db-ablation" - allowed_tools = ["skills_list", "skill_view", "terminal"] + allowed_tools = ["skills_list", "skill_view", STRUCTURED_READ_TOOL_NAME] toolsets.TOOLSETS[toolset_name] = { "description": "Fail-closed Leo OOS tool surface", "tools": allowed_tools, "includes": [], } tools_config._get_platform_tools = lambda *_args, **_kwargs: {toolset_name} - missing = sorted(name for name in allowed_tools if name not in registry._tools) + missing = sorted(name for name in ("skills_list", "skill_view") if name not in registry._tools) if missing: raise ReadOnlyGuardError(f"required tools are not registered: {missing}") - allowed_entries = {name: registry._tools[name] for name in allowed_tools} + allowed_entries = {name: registry._tools[name] for name in ("skills_list", "skill_view")} registry._tools.clear() registry._tools.update(allowed_entries) - wrapper = temp_profile / "bin" / "teleo-kb" - terminal_entry = registry._tools["terminal"] - terminal_entry.handler = lambda tool_args, **kwargs: restricted_bridge_terminal_handler( - wrapper, - temp_profile, - tool_args, - allow_kb_reads=allow_kb_reads, - **kwargs, + kb_tool = _verified_kb_tool(temp_profile, expected_kb_tool_sha256) + registry.register( + name=STRUCTURED_READ_TOOL_NAME, + toolset=toolset_name, + schema=STRUCTURED_READ_TOOL_SCHEMA, + handler=lambda tool_args, **kwargs: restricted_structured_read_handler( + kb_tool, + temp_profile, + tool_args, + allow_kb_reads=allow_kb_reads, + expected_kb_tool_sha256=expected_kb_tool_sha256, + **kwargs, + ), + description=STRUCTURED_READ_TOOL_SCHEMA["description"], ) + definitions = registry.get_definitions({STRUCTURED_READ_TOOL_NAME}, quiet=True) + if len(definitions) != 1: + raise ReadOnlyGuardError("structured KB read tool definition is not model-visible") + definition_sha256 = _canonical_sha256(definitions[0]) return { "mode": "read_only_kb" if allow_kb_reads else "no_db_ablation", "allowed_tools": allowed_tools, "actual_registry_tools": sorted(registry._tools), "read_only_bridge_commands": sorted(READ_ONLY_BRIDGE_COMMANDS) if allow_kb_reads else [], "send_message_tool_enabled": "send_message" in registry._tools, - "terminal_restricted_to_exact_wrapper": True, + "shell_tool_enabled": "terminal" in registry._tools, + "structured_read_tool_name": STRUCTURED_READ_TOOL_NAME, + "structured_read_tool_definition_sha256": definition_sha256, + "structured_read_tool_restricted_to_frozen_kb_tool": True, + "kb_tool_sha256": expected_kb_tool_sha256, + "bridge_timeout_seconds": BRIDGE_TIMEOUT_SECONDS, "mutating_bridge_commands_exposed": False, - "provider_credentials_forwarded_to_terminal": False, + "provider_credentials_forwarded_to_kb_tool": False, } diff --git a/scripts/leo_tool_trace.py b/scripts/leo_tool_trace.py index 986ee82..68fdd7c 100644 --- a/scripts/leo_tool_trace.py +++ b/scripts/leo_tool_trace.py @@ -54,6 +54,15 @@ ERROR_RE = re.compile( r"no such file or directory|connection (?:refused|failed))", re.IGNORECASE, ) +SAFE_ERROR_CATEGORIES = frozenset( + { + "bridge_exit", + "bridge_launch", + "bridge_timeout", + "invalid_read_arguments", + "kb_reads_disabled", + } +) def _sha256_bytes(value: bytes) -> str: @@ -64,6 +73,12 @@ def _canonical_bytes(value: Any) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") +def tool_arguments_sha256(value: Any) -> str: + """Hash typed tool arguments exactly as the retained trace does.""" + + return _sha256_bytes(_canonical_bytes(value)) + + def _normalize_tool_call(tool_call: Any) -> tuple[str | None, str, Any]: if not isinstance(tool_call, dict): return None, "", None @@ -131,7 +146,7 @@ def _database_invocations(tool_name: str, arguments: Any) -> list[dict[str, Any] if subcommand in MUTATING_SUBCOMMANDS else "unknown_write_risk" ), - "command_sha256": _sha256_bytes(_canonical_bytes(parsed_arguments)), + "command_sha256": tool_arguments_sha256(parsed_arguments), } ) return invocations @@ -145,20 +160,38 @@ def _sanitize_result(content: Any) -> dict[str, Any]: structured = json.loads(content) except json.JSONDecodeError: structured = None + payload_text = text structured_error = False + safe_exit_code = None + safe_error_category = None if isinstance(structured, dict): + if "output" in structured: + output = structured.get("output") + if isinstance(output, str): + payload_text = output + elif output is None: + payload_text = "" + else: + payload_text = json.dumps(output, sort_keys=True, default=str) exit_code = structured.get("exit_code") + if isinstance(exit_code, int) and not isinstance(exit_code, bool) and -255 <= exit_code <= 255: + safe_exit_code = exit_code + category = structured.get("error_category") + if isinstance(category, str) and category in SAFE_ERROR_CATEGORIES: + safe_error_category = category structured_error = bool( (isinstance(exit_code, int) and not isinstance(exit_code, bool) and exit_code != 0) or structured.get("error") ) - encoded = text.encode("utf-8") - uuids = sorted(set(UUID_RE.findall(text))) - hashes = sorted({value.lower() for value in SHA256_RE.findall(text)}) - schema_match = RECEIPT_SCHEMA_RE.search(text) - semantic_match = SEMANTIC_SHA_RE.search(text) - artifact_match = ARTIFACT_SHA_RE.search(text) - consistency_match = CONSISTENCY_RE.search(text) + if structured_error and safe_error_category is None: + safe_error_category = "structured_tool_error" + encoded = payload_text.encode("utf-8") + uuids = sorted(set(UUID_RE.findall(payload_text))) + hashes = sorted({value.lower() for value in SHA256_RE.findall(payload_text)}) + schema_match = RECEIPT_SCHEMA_RE.search(payload_text) + semantic_match = SEMANTIC_SHA_RE.search(payload_text) + artifact_match = ARTIFACT_SHA_RE.search(payload_text) + consistency_match = CONSISTENCY_RE.search(payload_text) retrieval_receipt = None if schema_match or semantic_match or artifact_match or consistency_match: retrieval_receipt = { @@ -170,8 +203,10 @@ def _sanitize_result(content: Any) -> dict[str, Any]: return { "content_sha256": _sha256_bytes(encoded), "content_bytes": len(encoded), - "nonempty": bool(text.strip()), - "error_detected": bool(structured_error or ERROR_RE.search(text)), + "nonempty": bool(payload_text.strip()), + "error_detected": bool(structured_error or ERROR_RE.search(payload_text)), + "exit_code": safe_exit_code, + "error_category": safe_error_category, "row_ids": uuids[:64], "row_id_count": len(uuids), "sha256_values": hashes[:64], @@ -199,7 +234,7 @@ def extract_kb_tool_trace(messages: list[dict[str, Any]] | None) -> dict[str, An "call_index": call_index, "tool_call_id_sha256": _sha256_bytes(raw_id_text.encode("utf-8")), "tool_name": tool_name, - "arguments_sha256": _sha256_bytes(_canonical_bytes(_parse_arguments(arguments))), + "arguments_sha256": tool_arguments_sha256(_parse_arguments(arguments)), "database_invocations": invocations, "result": None, } diff --git a/scripts/run_leo_m3taversal_oos_handler_suite.py b/scripts/run_leo_m3taversal_oos_handler_suite.py index 13b2030..f91e10f 100755 --- a/scripts/run_leo_m3taversal_oos_handler_suite.py +++ b/scripts/run_leo_m3taversal_oos_handler_suite.py @@ -176,6 +176,7 @@ def build_guarded_remote_script( + json.dumps(grounding_mode) + "\nOOS_GUARD_SOURCE_SHA256 = " + json.dumps(hashlib.sha256(guard_source.encode()).hexdigest()) + + "\nOOS_KB_TOOL_SOURCE_SHA256 = hashlib.sha256(KB_TOOL_SOURCE.encode()).hexdigest()" + "\nOOS_PROMPT_LEAKAGE_SCAN = " + repr(leakage_scan) + "\n\n" @@ -208,6 +209,7 @@ def build_guarded_remote_script( tool_surface = guard_module.install_read_only_tool_surface( temp_profile, allow_kb_reads=OOS_GROUNDING_MODE == "grounded", + expected_kb_tool_sha256=OOS_KB_TOOL_SOURCE_SHA256, ) report["executed_behavior_manifest"] = build_behavior_manifest(temp_profile) telegram_transport_attempts = [] @@ -355,9 +357,20 @@ def build_guarded_remote_script( report["oos_max_turn_seconds"] = 180 report["preexecution_safety_gate"] = { "status": "pass" if ( - tool_surface.get("send_message_tool_enabled") is False + tool_surface.get("actual_registry_tools") == ["skill_view", "skills_list", "teleo-kb"] + and tool_surface.get("allowed_tools") == ["skills_list", "skill_view", "teleo-kb"] + and tool_surface.get("send_message_tool_enabled") is False + and tool_surface.get("shell_tool_enabled") is False and tool_surface.get("mutating_bridge_commands_exposed") is False - and tool_surface.get("terminal_restricted_to_exact_wrapper") is True + and tool_surface.get("structured_read_tool_name") == "teleo-kb" + and tool_surface.get("structured_read_tool_restricted_to_frozen_kb_tool") is True + and re.fullmatch( + r"[0-9a-f]{64}", str(tool_surface.get("structured_read_tool_definition_sha256") or "") + ) is not None + and tool_surface.get("kb_tool_sha256") == OOS_KB_TOOL_SOURCE_SHA256 + and isinstance(tool_surface.get("bridge_timeout_seconds"), int) + and not isinstance(tool_surface.get("bridge_timeout_seconds"), bool) + and 0 < tool_surface.get("bridge_timeout_seconds") <= 60 and OOS_PROMPT_LEAKAGE_SCAN.get("pass") is True and report["remote_temp_profile_prompt_leakage_scan"]["pass"] is True and report["telegram_transport_deny"]["enabled"] is True @@ -384,6 +397,26 @@ def build_guarded_remote_script( script = script.replace( runner_marker, runner_marker + + " from tools.registry import registry as post_runner_registry\n" + + " post_runner_registry_tools = sorted(post_runner_registry._tools)\n" + + " post_runner_definitions = post_runner_registry.get_definitions({'teleo-kb'}, quiet=True)\n" + + " post_runner_definition_sha256 = (\n" + + " guard_module._canonical_sha256(post_runner_definitions[0])\n" + + " if len(post_runner_definitions) == 1 else None\n" + + " )\n" + + " post_runner_tool_surface_unchanged = bool(\n" + + " post_runner_registry_tools == ['skill_view', 'skills_list', 'teleo-kb']\n" + + " and post_runner_definition_sha256\n" + + " == tool_surface.get('structured_read_tool_definition_sha256')\n" + + " )\n" + + " report['post_runner_tool_surface'] = {\n" + + " 'actual_registry_tools': post_runner_registry_tools,\n" + + " 'structured_read_tool_definition_sha256': post_runner_definition_sha256,\n" + + " 'unchanged_from_preexecution': post_runner_tool_surface_unchanged,\n" + + " }\n" + + " write_report_snapshot(report)\n" + + " if not post_runner_tool_surface_unchanged:\n" + + " raise RuntimeError('OOS post-runner tool surface changed')\n" + " runner.adapters.clear()\n" + " runner.delivery_router.adapters = runner.adapters\n" + " report['telegram_transport_deny']['runner_adapters_empty'] = not runner.adapters\n" diff --git a/scripts/working_leo_m3taversal_oos_protocol.py b/scripts/working_leo_m3taversal_oos_protocol.py index f7bc053..c5d6de0 100644 --- a/scripts/working_leo_m3taversal_oos_protocol.py +++ b/scripts/working_leo_m3taversal_oos_protocol.py @@ -19,17 +19,18 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any +import leo_tool_trace as tool_trace_lib import leo_turn_execution_manifest as execution_manifest_lib import working_leo_m3taversal_oos_benchmark as benchmark ROOT = Path(__file__).resolve().parents[1] -PROTOCOL_SCHEMA = "livingip.leoM3taversalOosProtocol.v2" -TRIAL_SCORE_SCHEMA = "livingip.leoM3taversalOosTrialScore.v2" -AGGREGATE_SCHEMA = "livingip.leoM3taversalOosAggregate.v2" -GENERATOR_VERSION = "blinded-family-generator-v3" -SCORER_VERSION = "invariant-reasoning-live-receipts-and-factual-ablation-v5" -BASELINE_VERSION = "live-current-build-db-tool-ablation-v2" +PROTOCOL_SCHEMA = "livingip.leoM3taversalOosProtocol.v3" +TRIAL_SCORE_SCHEMA = "livingip.leoM3taversalOosTrialScore.v3" +AGGREGATE_SCHEMA = "livingip.leoM3taversalOosAggregate.v3" +GENERATOR_VERSION = "blinded-family-generator-v5" +SCORER_VERSION = "invariant-reasoning-live-receipts-and-factual-ablation-v7" +BASELINE_VERSION = "live-current-build-db-tool-ablation-v3" DEFAULT_TRIAL_COUNT = 3 MEMORY_SCORER_IDS = frozenset({"OOS-07", "OOS-08"}) DATABASE_CONTRACT_FAMILIES = frozenset( @@ -301,20 +302,23 @@ BLINDED_FAMILIES: tuple[dict[str, Any], ...] = ( "expected_follow_up": "return a model-visible live-read receipt token without mutation", "variants": ( ( - "For the {subject}, use the terminal tool to run `teleo-kb context '{subject}' --limit 1 " - "--context-limit 1 --format markdown` exactly once. Return four short lines: `Subject: {subject}`, " + "For the {subject}, use the `teleo-kb` tool exactly once with `action: context`, " + "`query: {subject}`, `limit: 1`, `context_limit: 1`, and `format: markdown`. Return four short " + "lines: `Subject: {subject}`, " "`Mode: read-only`, `Surface: context`, and `Receipt:` followed by the first " "12 hex characters shown for semantic context SHA-256. Do not infer or write anything." ), ( - "Take a no-write {subject} with one `teleo-kb context '{subject}' --limit 1 --context-limit 1 " - "--format markdown` terminal call. Answer only with `Subject: {subject}`, `Mode: read-only`, " + "Take a no-write {subject} with one `teleo-kb` tool call using `action: context`, " + "`query: {subject}`, `limit: 1`, `context_limit: 1`, and `format: markdown`. Answer only with " + "`Subject: {subject}`, `Mode: read-only`, " "`Surface: context`, and `Receipt: `." ), ( - "Prove the {subject} came from a live read: call `teleo-kb context '{subject}' --limit 1 " - "--context-limit 1 --format markdown` once, then give `Subject: {subject}`, `Mode: read-only`, " + "Prove the {subject} came from a live read: call the `teleo-kb` tool once with `action: context`, " + "`query: {subject}`, `limit: 1`, `context_limit: 1`, and `format: markdown`, then give " + "`Subject: {subject}`, `Mode: read-only`, " "`Surface: context`, and the first 12 semantic-context hash characters as " "`Receipt:`. No staging, apply, or other command." ), @@ -465,6 +469,18 @@ def _memory_token(seed: str, trial_index: int) -> str: return "blind-ledger-" + hashlib.sha256(f"{seed}:memory:{trial_index}".encode()).hexdigest()[:12] +def _expected_tool_arguments(subject: str, *, required: bool) -> dict[str, Any] | None: + if not required: + return None + return { + "action": "context", + "query": subject, + "limit": 1, + "context_limit": 1, + "format": "markdown", + } + + def build_blinded_trial(seed: str, trial_index: int, *, session_mode: str) -> dict[str, Any]: if not seed: raise ValueError("seed must not be empty") @@ -483,11 +499,7 @@ def build_blinded_trial(seed: str, trial_index: int, *, session_mode: str) -> di requires_database_receipt = family["family_id"] in DATABASE_RECEIPT_FAMILIES requires_tool_evidence = family["family_id"] == "receipt_discrimination" requires_grounded_retrieval_answer = family["family_id"] in AUTONOMOUS_RETRIEVAL_FAMILIES - expected_tool_command = ( - f"teleo-kb context '{subject}' --limit 1 --context-limit 1 --format markdown" - if requires_tool_evidence - else None - ) + expected_tool_arguments = _expected_tool_arguments(subject, required=requires_tool_evidence) message = variants[variant_index].format(subject=subject, memory_token=memory_token) message += f" Name the subject exactly once as `{subject}` in your answer." prompt_id = f"BLIND-{family['family_id'].upper()}-T{trial_index + 1:02d}-V{variant_index + 1:02d}" @@ -516,8 +528,8 @@ def build_blinded_trial(seed: str, trial_index: int, *, session_mode: str) -> di "requires_tool_evidence_token": requires_tool_evidence, "requires_grounded_retrieval_answer": requires_grounded_retrieval_answer, "custom_evidence_probe": family["scorer_id"] == "EVIDENCE-01", - "expected_tool_command_sha256": hashlib.sha256(expected_tool_command.encode()).hexdigest() - if expected_tool_command + "expected_tool_arguments_sha256": tool_trace_lib.tool_arguments_sha256(expected_tool_arguments) + if expected_tool_arguments else None, } ) @@ -558,14 +570,14 @@ def freeze_protocol( "same_model_profile_and_tool_schema": True, "ablated_surfaces": [ "temporary_profile.plugins.leo-db-context", - "successful teleo-kb terminal execution", + "successful structured teleo-kb read execution", ], "preserved_surfaces": [ "prompt manifest and order", "scorer and thresholds", "deployed build and model configuration", "temporary profile seed", - "model-visible skills and terminal tool schema", + "model-visible skills and structured teleo-kb tool schema", ], "expected_outcome": ( "zero successful DB receipts plus a lower factual answer score when both arms are checked against " @@ -654,19 +666,19 @@ def validate_protocol(protocol: dict[str, Any], *, verify_source_hashes: bool) - if requires_grounded_answer != (prompt.get("family_id") in AUTONOMOUS_RETRIEVAL_FAMILIES): issues.append(f"grounded_retrieval_answer_requirement_mismatch:{prompt_id}") if requires_grounded_answer and ( - "teleo-kb context" in message or "--limit" in message or "--context-limit" in message + "`teleo-kb` tool" in message or "`action: context`" in message or "`context_limit:" in message ): issues.append(f"autonomous_retrieval_exact_command_leak:{prompt_id}") - if requires_tool_evidence and ("teleo-kb context" not in message or "`Receipt:" not in message): + if requires_tool_evidence and ( + "`teleo-kb`" not in message or "`action: context`" not in message or "`Receipt:" not in message + ): issues.append(f"tool_evidence_instruction_missing:{prompt_id}") - expected_command = ( - f"teleo-kb context '{subject}' --limit 1 --context-limit 1 --format markdown" - if requires_tool_evidence - else None + expected_arguments = _expected_tool_arguments(subject, required=requires_tool_evidence) + expected_arguments_hash = ( + tool_trace_lib.tool_arguments_sha256(expected_arguments) if expected_arguments else None ) - expected_command_hash = hashlib.sha256(expected_command.encode()).hexdigest() if expected_command else None - if prompt.get("expected_tool_command_sha256") != expected_command_hash: - issues.append(f"tool_command_hash_mismatch:{prompt_id}") + if prompt.get("expected_tool_arguments_sha256") != expected_arguments_hash: + issues.append(f"tool_arguments_hash_mismatch:{prompt_id}") family_id = str(prompt.get("family_id") or "") if family_id in family_by_id and prompt.get("family_subjects") != list(family_by_id[family_id]["subjects"]): issues.append(f"family_subjects_mismatch:{prompt_id}") @@ -706,14 +718,14 @@ def _subject_alignment(prompt: dict[str, Any], reply: str) -> bool: ) -def _tool_evidence_hashes(result: dict[str, Any], *, expected_command_sha256: str | None) -> list[str]: +def _tool_evidence_hashes(result: dict[str, Any], *, expected_arguments_sha256: str | None) -> list[str]: trace = result.get("database_tool_trace") if not isinstance(trace, dict) or trace.get("schema") != "livingip.leoKbToolTrace.v1": return [] hashes: set[str] = set() calls = trace.get("calls") if isinstance(trace.get("calls"), list) else [] if ( - not _valid_sha256(expected_command_sha256) + not _valid_sha256(expected_arguments_sha256) or len(calls) != 1 or trace.get("database_tool_call_count") != 1 or trace.get("database_tool_completed_count") != 1 @@ -737,7 +749,8 @@ def _tool_evidence_hashes(result: dict[str, Any], *, expected_command_sha256: st len(invocations) != 1 or invocations[0].get("executable") != "teleo-kb" or invocations[0].get("subcommand") != "context" - or invocations[0].get("command_sha256") != expected_command_sha256 + or call.get("arguments_sha256") != expected_arguments_sha256 + or invocations[0].get("command_sha256") != expected_arguments_sha256 ): continue receipt = result_summary.get("retrieval_receipt") @@ -1337,6 +1350,8 @@ def _top_level_safety( expected_harness_git_head=expected_harness_git_head, ) tool_surface = report.get("read_only_tool_surface") or {} + post_runner_tool_surface = report.get("post_runner_tool_surface") or {} + report_source_hashes = report.get("source_hashes") or {} handler_safety = report.get("safety_gate") or {} orphan_readback = report.get("post_run_orphan_readback") or {} leakage_scan = report.get("prompt_leakage_scan") or {} @@ -1380,12 +1395,30 @@ def _top_level_safety( "temporary_profile_removed": report.get("temp_profile_removed") is True, "execution_chain_complete_under_declared_benchmark_mode": benchmark_execution["pass"], "tool_registry_exactly_allowlisted": tool_surface.get("actual_registry_tools") - == ["skill_view", "skills_list", "terminal"], + == ["skill_view", "skills_list", "teleo-kb"], "send_message_tool_absent": tool_surface.get("send_message_tool_enabled") is False, + "shell_tool_absent": tool_surface.get("shell_tool_enabled") is False, "mutating_bridge_commands_not_exposed": tool_surface.get("mutating_bridge_commands_exposed") is False, - "terminal_provider_credentials_not_forwarded": tool_surface.get("provider_credentials_forwarded_to_terminal") + "kb_tool_provider_credentials_not_forwarded": tool_surface.get("provider_credentials_forwarded_to_kb_tool") is False, - "terminal_restricted_to_exact_wrapper": tool_surface.get("terminal_restricted_to_exact_wrapper") is True, + "structured_read_tool_named": tool_surface.get("structured_read_tool_name") == "teleo-kb", + "structured_read_tool_restricted_to_frozen_kb_tool": tool_surface.get( + "structured_read_tool_restricted_to_frozen_kb_tool" + ) + is True, + "structured_read_tool_schema_hash_bound": _valid_sha256( + tool_surface.get("structured_read_tool_definition_sha256") + ), + "structured_read_kb_tool_hash_bound": _valid_sha256(tool_surface.get("kb_tool_sha256")) + and tool_surface.get("kb_tool_sha256") == report_source_hashes.get("kb_tool_sha256"), + "structured_read_timeout_bounded": isinstance(tool_surface.get("bridge_timeout_seconds"), int) + and not isinstance(tool_surface.get("bridge_timeout_seconds"), bool) + and 0 < tool_surface.get("bridge_timeout_seconds") <= 60, + "post_runner_tool_registry_still_exact": post_runner_tool_surface.get("actual_registry_tools") + == ["skill_view", "skills_list", "teleo-kb"], + "post_runner_tool_schema_unchanged": post_runner_tool_surface.get("unchanged_from_preexecution") is True + and post_runner_tool_surface.get("structured_read_tool_definition_sha256") + == tool_surface.get("structured_read_tool_definition_sha256"), "handler_safety_gate_passed_or_only_declared_manifest_gap": handler_gate_acceptable if require_handler_safety_gate else True, @@ -1698,7 +1731,7 @@ def score_live_trial( result_item = report_by_prompt.get(prompt_id) or {} tool_evidence_hashes = _tool_evidence_hashes( result_item, - expected_command_sha256=prompt.get("expected_tool_command_sha256"), + expected_arguments_sha256=prompt.get("expected_tool_arguments_sha256"), ) evidence_answer = _evidence_answer_score( prompt, @@ -1928,6 +1961,8 @@ def score_live_trial( return [identities[key] for key in sorted(identities)] executed_behavior_ablation = _executed_behavior_ablation(report, baseline_report) + baseline_tool_surface = baseline_report.get("read_only_tool_surface") or {} + grounded_tool_surface = report.get("read_only_tool_surface") or {} comparison_axis_checks = { "protocol_id_equal": baseline_report.get("protocol_id") == report.get("protocol_id") == protocol["protocol_id"], "protocol_hash_equal": baseline_report.get("protocol_hash_sha256") @@ -1955,8 +1990,13 @@ def score_live_trial( "readonly_guard_source_equal_and_frozen": baseline_report.get("readonly_guard_source_sha256") == report.get("readonly_guard_source_sha256") == protocol["source_hashes"]["readonly_guard_sha256"], - "tool_schema_equal": (baseline_report.get("read_only_tool_surface") or {}).get("allowed_tools") - == (report.get("read_only_tool_surface") or {}).get("allowed_tools"), + "tool_allowlist_equal": baseline_tool_surface.get("allowed_tools") + == grounded_tool_surface.get("allowed_tools"), + "tool_schema_equal": _valid_sha256(baseline_tool_surface.get("structured_read_tool_definition_sha256")) + and baseline_tool_surface.get("structured_read_tool_definition_sha256") + == grounded_tool_surface.get("structured_read_tool_definition_sha256"), + "frozen_kb_tool_equal": _valid_sha256(baseline_tool_surface.get("kb_tool_sha256")) + and baseline_tool_surface.get("kb_tool_sha256") == grounded_tool_surface.get("kb_tool_sha256"), } threshold = float(protocol["thresholds"]["minimum_trial_grounded_pass_rate"]) evidence_threshold = float(protocol["thresholds"]["minimum_trial_evidence_answer_pass_rate"]) diff --git a/tests/test_leo_oos_readonly_guard.py b/tests/test_leo_oos_readonly_guard.py index e17b2c2..3265226 100644 --- a/tests/test_leo_oos_readonly_guard.py +++ b/tests/test_leo_oos_readonly_guard.py @@ -2,9 +2,12 @@ from __future__ import annotations +import hashlib import json +import signal import subprocess import sys +import types from pathlib import Path import pytest @@ -15,51 +18,29 @@ sys.path.insert(0, str(ROOT / "scripts")) import leo_oos_readonly_guard as guard # noqa: E402 -def make_wrapper(tmp_path: Path) -> tuple[Path, Path]: +def make_kb_tool(tmp_path: Path) -> tuple[Path, Path, str]: profile = tmp_path / "profile" - wrapper = profile / "bin" / "teleo-kb" - wrapper.parent.mkdir(parents=True) - wrapper.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") - wrapper.chmod(0o700) - return profile, wrapper + kb_tool = profile / "bin" / "kb_tool.py" + kb_tool.parent.mkdir(parents=True) + kb_tool.write_text("#!/usr/bin/env python3\n", encoding="utf-8") + kb_tool.chmod(0o700) + return profile, kb_tool.resolve(), hashlib.sha256(kb_tool.read_bytes()).hexdigest() -@pytest.mark.parametrize( - "command", - [ - "teleo-kb status --format json", - "teleo-kb search 'Orchid forecast' --limit 5 --context-limit 4 --format json", - "teleo-kb list-proposals --status approved --limit 10 --format markdown", - "teleo-kb show 11111111-1111-4111-8111-111111111111 --format json", - ], -) -def test_guard_accepts_only_exact_read_only_bridge_argv(tmp_path: Path, command: str) -> None: - profile, wrapper = make_wrapper(tmp_path) - argv, reason = guard.classify_terminal_command( - wrapper, - profile, - {"command": command}, - allow_kb_reads=True, - ) - assert argv is not None - assert reason is None - - -def test_guard_accepts_unquoted_multiword_read_query_as_one_safe_invocation(tmp_path: Path) -> None: - profile, wrapper = make_wrapper(tmp_path) - argv, reason = guard.classify_terminal_command( - wrapper, - profile, - {"command": "teleo-kb context Atlas status snapshot --limit 1 --context-limit 1 --format markdown"}, - allow_kb_reads=True, +def test_structured_context_compiles_to_one_exact_read_only_argv() -> None: + argv = guard.structured_read_argv( + { + "action": "context", + "query": "Northstar status snapshot", + "limit": 1, + "context_limit": 1, + "format": "markdown", + } ) assert argv == [ - "teleo-kb", "context", - "Atlas", - "status", - "snapshot", + "Northstar status snapshot", "--limit", "1", "--context-limit", @@ -67,74 +48,292 @@ def test_guard_accepts_unquoted_multiword_read_query_as_one_safe_invocation(tmp_ "--format", "markdown", ] - assert reason is None + + +def test_structured_read_omits_unsupplied_flags_so_bridge_defaults_remain_canonical() -> None: + assert guard.structured_read_argv({"action": "evidence", "claim_id": "11111111-1111-4111-8111-111111111111"}) == [ + "evidence", + "11111111-1111-4111-8111-111111111111", + ] @pytest.mark.parametrize( "tool_args", [ - {"command": "teleo-kb propose-source /tmp/input"}, - {"command": "teleo-kb record-document-evaluation /tmp/input"}, - {"command": "teleo-kb status; rm -rf /tmp/x"}, - {"command": "bash -lc 'teleo-kb status'"}, - {"command": "teleo-kb status", "background": True}, - {"command": "teleo-kb status", "pty": True}, - {"command": "teleo-kb status", "workdir": "/tmp"}, - {"command": "teleo-kb evidence not-a-uuid"}, - {"command": "teleo-kb search x --limit 1000"}, + {"action": "prepare-source", "query": "x"}, + {"action": "context", "query": "x", "rationale": "write it"}, + {"action": "context"}, + {"action": "context", "query": "x", "limit": 0}, + {"action": "context", "query": "x", "context_limit": True}, + {"action": "context", "query": "x", "format": []}, + {"action": "show", "claim_id": "not-a-uuid"}, + {"action": "list-proposals", "status": []}, + {"action": "status", "query": "irrelevant"}, ], ) -def test_guard_rejects_writes_shell_escape_and_unbounded_arguments( - tmp_path: Path, tool_args: dict[str, object] +def test_structured_read_rejects_writes_bad_types_irrelevant_fields_and_unbounded_values( + tool_args: dict[str, object], ) -> None: - profile, wrapper = make_wrapper(tmp_path) - argv, reason = guard.classify_terminal_command( - wrapper, - profile, - tool_args, - allow_kb_reads=True, - ) - assert argv is None - assert reason + with pytest.raises(guard.ReadOnlyGuardError): + guard.structured_read_argv(tool_args) -def test_no_db_ablation_preserves_terminal_schema_but_denies_execution(tmp_path: Path) -> None: - profile, wrapper = make_wrapper(tmp_path) - argv, reason = guard.classify_terminal_command( - wrapper, - profile, - {"command": "teleo-kb status --format json"}, - allow_kb_reads=False, - ) - assert argv is None - assert reason == "database tools are disabled for the no-DB ablation" +def test_schema_encodes_action_specific_required_fields() -> None: + branches = guard.STRUCTURED_READ_TOOL_SCHEMA["parameters"]["oneOf"] + by_action = {branch["properties"]["action"]["enum"][0]: branch for branch in branches} + + assert by_action["context"]["required"] == ["action", "query"] + assert set(by_action["context"]["properties"]) == {"action", "query", "limit", "context_limit", "format"} + assert by_action["context"]["additionalProperties"] is False + assert by_action["show"]["required"] == ["action", "claim_id"] + assert by_action["status"]["required"] == ["action"] -def test_terminal_child_uses_temp_profile_and_no_provider_credentials( +class FakeProcess: + def __init__(self, argv, **kwargs): + self.argv = argv + self.kwargs = kwargs + self.pid = 8123 + self.returncode = 0 + + def communicate(self, timeout=None): + return "{}\n", "" + + +def test_structured_read_child_uses_frozen_absolute_tool_and_no_provider_credentials( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: - profile, wrapper = make_wrapper(tmp_path) + profile, kb_tool, source_sha256 = make_kb_tool(tmp_path) captured: dict[str, object] = {} - def fake_run(argv, **kwargs): - captured["argv"] = argv - captured["env"] = kwargs["env"] - return subprocess.CompletedProcess(argv, 0, stdout="{}\n", stderr="") + def fake_popen(argv, **kwargs): + proc = FakeProcess(argv, **kwargs) + captured["proc"] = proc + return proc - monkeypatch.setattr(guard.subprocess, "run", fake_run) + monkeypatch.setattr(guard.subprocess, "Popen", fake_popen) output = json.loads( - guard.restricted_bridge_terminal_handler( - wrapper, + guard.restricted_structured_read_handler( + kb_tool, profile, - {"command": "teleo-kb status --format json"}, + { + "action": "context", + "query": "Northstar status snapshot", + "limit": 1, + "context_limit": 1, + "format": "markdown", + }, allow_kb_reads=True, + expected_kb_tool_sha256=source_sha256, ) ) - assert output["exit_code"] == 0 - assert captured["env"] == { + + assert output == {"output": "{}\n", "exit_code": 0, "error": None, "error_category": None} + proc = captured["proc"] + assert proc.argv == [ + str(Path(sys.executable).resolve()), + str(kb_tool), + "--local", + "context", + "Northstar status snapshot", + "--limit", + "1", + "--context-limit", + "1", + "--format", + "markdown", + ] + assert proc.kwargs["start_new_session"] is True + assert proc.kwargs["env"] == { "HOME": str(profile), "HERMES_HOME": str(profile), "PATH": f"{profile / 'bin'}:{guard.os.environ.get('PATH', '')}", "TELEO_KB_MODE": "local", } - assert not any("KEY" in key or "TOKEN" in key or "SECRET" in key for key in captured["env"]) + assert not any("KEY" in key or "TOKEN" in key or "SECRET" in key for key in proc.kwargs["env"]) + + +def test_structured_read_rejects_hash_drift_before_launch(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + profile, kb_tool, source_sha256 = make_kb_tool(tmp_path) + kb_tool.write_text("changed\n", encoding="utf-8") + monkeypatch.setattr( + guard.subprocess, + "Popen", + lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("must not launch")), + ) + + output = json.loads( + guard.restricted_structured_read_handler( + kb_tool, + profile, + {"action": "status"}, + allow_kb_reads=True, + expected_kb_tool_sha256=source_sha256, + ) + ) + + assert output["exit_code"] == 126 + assert output["error_category"] == "invalid_read_arguments" + + +def test_structured_read_timeout_terminates_the_process_group(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + profile, kb_tool, source_sha256 = make_kb_tool(tmp_path) + calls = 0 + signals: list[tuple[int, signal.Signals]] = [] + + class TimeoutProcess(FakeProcess): + def communicate(self, timeout=None): + nonlocal calls + calls += 1 + if calls == 1: + raise subprocess.TimeoutExpired(self.argv, timeout) + return "", "" + + monkeypatch.setattr(guard.subprocess, "Popen", TimeoutProcess) + monkeypatch.setattr(guard.os, "killpg", lambda pid, sent: signals.append((pid, sent))) + output = json.loads( + guard.restricted_structured_read_handler( + kb_tool, + profile, + {"action": "status"}, + allow_kb_reads=True, + expected_kb_tool_sha256=source_sha256, + ) + ) + + assert output["exit_code"] == 124 + assert output["error_category"] == "bridge_timeout" + assert signals == [(8123, signal.SIGTERM)] + + +def test_structured_read_timeout_escalates_to_process_group_kill( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + profile, kb_tool, source_sha256 = make_kb_tool(tmp_path) + signals: list[tuple[int, signal.Signals]] = [] + + class StubbornProcess(FakeProcess): + def communicate(self, timeout=None): + raise subprocess.TimeoutExpired(self.argv, timeout) + + monkeypatch.setattr(guard.subprocess, "Popen", StubbornProcess) + monkeypatch.setattr(guard.os, "killpg", lambda pid, sent: signals.append((pid, sent))) + + output = json.loads( + guard.restricted_structured_read_handler( + kb_tool, + profile, + {"action": "status"}, + allow_kb_reads=True, + expected_kb_tool_sha256=source_sha256, + ) + ) + + assert output["exit_code"] == 124 + assert output["error_category"] == "bridge_timeout" + assert signals == [(8123, signal.SIGTERM), (8123, signal.SIGKILL)] + + +def test_structured_read_launch_error_is_categorized(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + profile, kb_tool, source_sha256 = make_kb_tool(tmp_path) + monkeypatch.setattr(guard.subprocess, "Popen", lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("nope"))) + + output = json.loads( + guard.restricted_structured_read_handler( + kb_tool, + profile, + {"action": "status"}, + allow_kb_reads=True, + expected_kb_tool_sha256=source_sha256, + ) + ) + + assert output["exit_code"] == 126 + assert output["error_category"] == "bridge_launch" + assert "nope" not in output["error"] + + +def test_structured_read_ablation_preserves_schema_but_never_executes( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + profile, kb_tool, source_sha256 = make_kb_tool(tmp_path) + monkeypatch.setattr( + guard.subprocess, + "Popen", + lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("ablation must not execute")), + ) + + output = json.loads( + guard.restricted_structured_read_handler( + kb_tool, + profile, + {"action": "status", "format": "json"}, + allow_kb_reads=False, + expected_kb_tool_sha256=source_sha256, + ) + ) + + assert output["exit_code"] == 126 + assert output["error_category"] == "kb_reads_disabled" + + +class FakeEntry: + def __init__(self, handler=None, schema=None): + self.handler = handler + self.schema = schema or {} + + +class FakeRegistry: + def __init__(self): + self._tools = {"skills_list": FakeEntry(), "skill_view": FakeEntry(), "terminal": FakeEntry()} + + def register(self, *, name, schema, handler, **_kwargs): + self._tools[name] = FakeEntry(handler=handler, schema=schema) + + def get_definitions(self, names, quiet=False): + del quiet + return [ + {"type": "function", "function": {**self._tools[name].schema, "name": name}} + for name in sorted(names) + if name in self._tools + ] + + def dispatch(self, name, args): + return self._tools[name].handler(args) + + +def test_registry_install_exposes_exact_schema_and_denies_ablation( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + profile, _kb_tool, source_sha256 = make_kb_tool(tmp_path) + registry = FakeRegistry() + toolsets = types.ModuleType("toolsets") + toolsets.TOOLSETS = {} + tools = types.ModuleType("tools") + tools.__path__ = [] + skills_tool = types.ModuleType("tools.skills_tool") + registry_module = types.ModuleType("tools.registry") + registry_module.registry = registry + hermes_cli = types.ModuleType("hermes_cli") + tools_config = types.SimpleNamespace(_get_platform_tools=lambda: set()) + hermes_cli.tools_config = tools_config + for name, module in { + "tools": tools, + "tools.skills_tool": skills_tool, + "tools.registry": registry_module, + "toolsets": toolsets, + "hermes_cli": hermes_cli, + }.items(): + monkeypatch.setitem(sys.modules, name, module) + + surface = guard.install_read_only_tool_surface( + profile, + allow_kb_reads=False, + expected_kb_tool_sha256=source_sha256, + ) + denied = json.loads(registry.dispatch("teleo-kb", {"action": "status"})) + + assert surface["actual_registry_tools"] == ["skill_view", "skills_list", "teleo-kb"] + assert surface["shell_tool_enabled"] is False + assert len(surface["structured_read_tool_definition_sha256"]) == 64 + assert denied["error_category"] == "kb_reads_disabled" diff --git a/tests/test_leo_tool_trace.py b/tests/test_leo_tool_trace.py index 78e644a..f78c7cf 100644 --- a/tests/test_leo_tool_trace.py +++ b/tests/test_leo_tool_trace.py @@ -9,7 +9,7 @@ from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "scripts")) -from leo_tool_trace import extract_kb_tool_trace # noqa: E402 +from leo_tool_trace import extract_kb_tool_trace, tool_arguments_sha256 # noqa: E402 def test_extracts_completed_read_only_context_call_and_receipt() -> None: @@ -164,6 +164,144 @@ def test_structured_nonzero_exit_does_not_prove_database_call() -> None: assert trace["database_tool_call_proven"] is False assert trace["database_tool_completed_count"] == 0 assert trace["calls"][0]["result"]["error_detected"] is True + assert trace["calls"][0]["result"]["exit_code"] == 2 + assert trace["calls"][0]["result"]["error_category"] == "structured_tool_error" + + +def test_structured_read_tool_is_attributed_without_retaining_raw_arguments() -> None: + arguments = { + "action": "context", + "query": "Northstar status snapshot", + "limit": 1, + "context_limit": 1, + "format": "markdown", + } + output = ( + "# Teleo KB Context\n" + "- schema: `livingip.teleoKbRetrievalReceipt.v1`\n" + f"- semantic context SHA-256: `{'a' * 64}`\n" + f"- artifact state SHA-256: `{'b' * 64}`\n" + "- DB read consistency: `stable_wal_marker`; attempts: `1`\n" + ) + messages = [ + { + "role": "assistant", + "tool_calls": [ + { + "id": "structured-read", + "function": { + "name": "teleo-kb", + "arguments": json.dumps(arguments), + }, + } + ], + }, + { + "role": "tool", + "tool_call_id": "structured-read", + "content": json.dumps({"output": output, "exit_code": 0, "error": None, "error_category": None}), + }, + ] + + trace = extract_kb_tool_trace(messages) + + assert trace["database_tool_call_count"] == 1 + assert trace["database_tool_completed_count"] == 1 + assert trace["database_retrieval_receipt_proven"] is True + assert trace["database_tool_calls_read_only"] is True + invocation = trace["calls"][0]["database_invocations"][0] + assert invocation["executable"] == "teleo-kb" + assert invocation["subcommand"] == "context" + assert trace["calls"][0]["arguments_sha256"] == invocation["command_sha256"] + assert trace["calls"][0]["result"]["content_bytes"] == len(output.encode()) + assert "Northstar status snapshot" not in str(trace) + + +def test_empty_success_envelope_does_not_prove_a_database_call() -> None: + arguments = {"action": "status", "format": "json"} + messages = [ + { + "role": "assistant", + "tool_calls": [ + { + "id": "empty-read", + "function": {"name": "teleo-kb", "arguments": json.dumps(arguments)}, + } + ], + }, + { + "role": "tool", + "tool_call_id": "empty-read", + "content": json.dumps({"output": "", "exit_code": 0, "error": None, "error_category": None}), + }, + ] + + trace = extract_kb_tool_trace(messages) + + assert trace["database_tool_call_count"] == 1 + assert trace["database_tool_completed_count"] == 0 + assert trace["database_tool_call_proven"] is False + assert trace["database_retrieval_receipt_proven"] is False + assert trace["calls"][0]["result"]["nonempty"] is False + + +def test_tool_argument_hash_is_shared_for_unicode_arguments() -> None: + arguments = {"action": "context", "query": "caf\u00e9", "limit": 1} + trace = extract_kb_tool_trace( + [ + { + "role": "assistant", + "tool_calls": [ + { + "id": "unicode-read", + "function": {"name": "teleo-kb", "arguments": json.dumps(arguments)}, + } + ], + }, + {"role": "tool", "tool_call_id": "unicode-read", "content": "ok"}, + ] + ) + + expected = tool_arguments_sha256(arguments) + assert trace["calls"][0]["arguments_sha256"] == expected + assert trace["calls"][0]["database_invocations"][0]["command_sha256"] == expected + + +def test_safe_structured_error_category_is_retained_without_raw_error() -> None: + messages = [ + { + "role": "assistant", + "tool_calls": [ + { + "id": "structured-error", + "function": { + "name": "teleo-kb", + "arguments": '{"action":"context","query":"secret query"}', + }, + } + ], + }, + { + "role": "tool", + "tool_call_id": "structured-error", + "content": json.dumps( + { + "output": "", + "exit_code": 126, + "error": "secret query was malformed", + "error_category": "invalid_read_arguments", + } + ), + }, + ] + + trace = extract_kb_tool_trace(messages) + + result = trace["calls"][0]["result"] + assert result["error_detected"] is True + assert result["exit_code"] == 126 + assert result["error_category"] == "invalid_read_arguments" + assert "secret query" not in str(trace) def test_actual_staging_subcommands_and_unknown_subcommands_fail_read_only_classification() -> None: diff --git a/tests/test_run_leo_m3taversal_oos_handler_suite.py b/tests/test_run_leo_m3taversal_oos_handler_suite.py index 772f283..2cf1400 100644 --- a/tests/test_run_leo_m3taversal_oos_handler_suite.py +++ b/tests/test_run_leo_m3taversal_oos_handler_suite.py @@ -48,6 +48,13 @@ def test_guarded_remote_script_compiles_and_installs_preexecution_gate(grounding assert "2_000_000" not in script assert 'raise RuntimeError("OOS pre-execution safety gate failed")' in script assert "send_message_tool_enabled" in script + assert "shell_tool_enabled" in script + assert 'structured_read_tool_name") == "teleo-kb"' in script + assert "structured_read_tool_restricted_to_frozen_kb_tool" in script + assert "expected_kb_tool_sha256=OOS_KB_TOOL_SOURCE_SHA256" in script + assert "structured_read_tool_definition_sha256" in script + assert "post_runner_tool_surface_unchanged" in script + assert "OOS post-runner tool surface changed" in script assert '"scope": "full_model_visible_temp_profile_excluding_sessions_state_memories_and_venv"' in script assert "Telegram transport is denied in the OOS no-post harness" in script assert "runner.adapters.clear()" in script diff --git a/tests/test_working_leo_m3taversal_oos_protocol.py b/tests/test_working_leo_m3taversal_oos_protocol.py index 56bdaeb..7bbc248 100644 --- a/tests/test_working_leo_m3taversal_oos_protocol.py +++ b/tests/test_working_leo_m3taversal_oos_protocol.py @@ -42,13 +42,18 @@ def frozen_protocol() -> dict: def tool_surface(mode: str) -> dict: return { "mode": mode, - "allowed_tools": ["skills_list", "skill_view", "terminal"], - "actual_registry_tools": ["skill_view", "skills_list", "terminal"], + "allowed_tools": ["skills_list", "skill_view", "teleo-kb"], + "actual_registry_tools": ["skill_view", "skills_list", "teleo-kb"], "read_only_bridge_commands": ["status"] if mode == "read_only_kb" else [], "send_message_tool_enabled": False, - "terminal_restricted_to_exact_wrapper": True, + "shell_tool_enabled": False, + "structured_read_tool_name": "teleo-kb", + "structured_read_tool_definition_sha256": "a" * 64, + "structured_read_tool_restricted_to_frozen_kb_tool": True, + "kb_tool_sha256": protocol_lib.file_sha256(protocol_lib.source_paths()["kb_tool_sha256"]), + "bridge_timeout_seconds": 45, "mutating_bridge_commands_exposed": False, - "provider_credentials_forwarded_to_terminal": False, + "provider_credentials_forwarded_to_kb_tool": False, } @@ -137,9 +142,10 @@ def result_for(prompt: dict, token: str, turn: int, *, grounded: bool) -> dict: "access_mode": "read_only", "executable": "teleo-kb", "subcommand": "context", - "command_sha256": prompt["expected_tool_command_sha256"], + "command_sha256": prompt["expected_tool_arguments_sha256"], } ], + "arguments_sha256": prompt["expected_tool_arguments_sha256"], "result": { "nonempty": True, "error_detected": False, @@ -462,6 +468,11 @@ def fake_report( }, "executed_behavior_manifest": fake_behavior_manifest(grounded=grounded), "read_only_tool_surface": tool_surface("read_only_kb" if grounded else "no_db_ablation"), + "post_runner_tool_surface": { + "actual_registry_tools": ["skill_view", "skills_list", "teleo-kb"], + "structured_read_tool_definition_sha256": "a" * 64, + "unchanged_from_preexecution": True, + }, "readonly_guard_source_sha256": protocol["source_hashes"]["readonly_guard_sha256"], "safety_gate": { "status": "fail", @@ -659,8 +670,8 @@ def test_protocol_freezes_three_unique_variants_and_follow_up_shapes() -> None: ] assert len({item["subject"] for item in retrieval_prompts}) == 3 assert all(item["requires_grounded_retrieval_answer"] is True for item in retrieval_prompts) - assert all("teleo-kb context" not in item["message"] for item in retrieval_prompts) - assert all("--limit" not in item["message"] for item in retrieval_prompts) + assert all("`teleo-kb` tool" not in item["message"] for item in retrieval_prompts) + assert all("`action: context`" not in item["message"] for item in retrieval_prompts) def test_protocol_validation_rejects_prompt_and_source_hash_tampering() -> None: @@ -1201,9 +1212,7 @@ def test_subject_alignment_rejects_a_different_family_even_with_generic_db_words ) assert protocol_lib._subject_alignment(retrieval_prompt, good + f" {retrieval_sibling}.") is False - memory_prompt = next( - item for item in protocol["trials"][0]["prompts"] if item["family_id"] == "session_memory_set" - ) + memory_prompt = next(item for item in protocol["trials"][0]["prompts"] if item["family_id"] == "session_memory_set") memory_reply = f"Subject: {memory_prompt['subject']}\nLabel: temporary\nBlocker: database schema gap." assert protocol_lib._subject_alignment(memory_prompt, memory_reply) is True @@ -1234,9 +1243,7 @@ def test_evidence_probe_four_line_reply_binds_exact_subject_and_receipt() -> Non def test_every_runtime_persistence_variant_declares_the_scorer_word_limit() -> None: - runtime_family = next( - item for item in protocol_lib.BLINDED_FAMILIES if item["family_id"] == "runtime_persistence" - ) + runtime_family = next(item for item in protocol_lib.BLINDED_FAMILIES if item["family_id"] == "runtime_persistence") assert all("under 180 words" in variant.lower() for variant in runtime_family["variants"])