diff --git a/scripts/leo_oos_readonly_guard.py b/scripts/leo_oos_readonly_guard.py index 23647f4..3569f34 100644 --- a/scripts/leo_oos_readonly_guard.py +++ b/scripts/leo_oos_readonly_guard.py @@ -16,6 +16,7 @@ from typing import Any STRUCTURED_READ_TOOL_NAME = "teleo-kb" BRIDGE_TIMEOUT_SECONDS = 45 BRIDGE_TERMINATE_GRACE_SECONDS = 5 +KNOWN_RUNNER_ADDED_TOOLS = frozenset({"process"}) PROPOSAL_STATUSES = ("pending_review", "approved", "applied", "canceled", "rejected", "all") _FIELD_SCHEMAS = { @@ -356,3 +357,43 @@ def install_read_only_tool_surface( "mutating_bridge_commands_exposed": False, "provider_credentials_forwarded_to_kb_tool": False, } + + +def reassert_read_only_tool_surface( + temp_profile: Path, + *, + expected_kb_tool_sha256: str, + expected_definition_sha256: str, +) -> dict[str, Any]: + """Remove the runner's known additions before the first model turn.""" + + from tools.registry import registry + + _verified_kb_tool(temp_profile, expected_kb_tool_sha256) + exact_tools = {"skills_list", "skill_view", STRUCTURED_READ_TOOL_NAME} + initial_tools = set(registry._tools) + missing = sorted(exact_tools - initial_tools) + unexpected = sorted(initial_tools - exact_tools) + unsupported = sorted(set(unexpected) - KNOWN_RUNNER_ADDED_TOOLS) + if missing or unsupported: + raise ReadOnlyGuardError( + f"post-runner registry mismatch; missing={missing}; unsupported additions={unsupported}" + ) + retained_entries = {name: registry._tools[name] for name in exact_tools} + registry._tools.clear() + registry._tools.update(retained_entries) + definitions = registry.get_definitions({STRUCTURED_READ_TOOL_NAME}, quiet=True) + if len(definitions) != 1: + raise ReadOnlyGuardError("structured KB read tool definition disappeared after runner initialization") + definition_sha256 = _canonical_sha256(definitions[0]) + if definition_sha256 != expected_definition_sha256: + raise ReadOnlyGuardError("structured KB read tool schema changed after runner initialization") + final_tools = sorted(registry._tools) + expected_final_tools = sorted(exact_tools) + return { + "initial_registry_tools": sorted(initial_tools), + "removed_known_runner_tools": unexpected, + "actual_registry_tools": final_tools, + "structured_read_tool_definition_sha256": definition_sha256, + "final_surface_matches_preexecution": final_tools == expected_final_tools, + } diff --git a/scripts/run_leo_m3taversal_oos_handler_suite.py b/scripts/run_leo_m3taversal_oos_handler_suite.py index f91e10f..c87a9a3 100755 --- a/scripts/run_leo_m3taversal_oos_handler_suite.py +++ b/scripts/run_leo_m3taversal_oos_handler_suite.py @@ -397,25 +397,16 @@ 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" + + " post_runner_tool_surface = guard_module.reassert_read_only_tool_surface(\n" + + " temp_profile,\n" + + " expected_kb_tool_sha256=OOS_KB_TOOL_SOURCE_SHA256,\n" + + " expected_definition_sha256=(\n" + + " tool_surface['structured_read_tool_definition_sha256']\n" + + " ),\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" + + " report['post_runner_tool_surface'] = post_runner_tool_surface\n" + " write_report_snapshot(report)\n" - + " if not post_runner_tool_surface_unchanged:\n" + + " if not post_runner_tool_surface.get('final_surface_matches_preexecution'):\n" + " raise RuntimeError('OOS post-runner tool surface changed')\n" + " runner.adapters.clear()\n" + " runner.delivery_router.adapters = runner.adapters\n" diff --git a/scripts/working_leo_m3taversal_oos_protocol.py b/scripts/working_leo_m3taversal_oos_protocol.py index c5d6de0..085f14f 100644 --- a/scripts/working_leo_m3taversal_oos_protocol.py +++ b/scripts/working_leo_m3taversal_oos_protocol.py @@ -1416,7 +1416,10 @@ def _top_level_safety( 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 + "post_runner_only_known_framework_tools_removed": set( + post_runner_tool_surface.get("removed_known_runner_tools") or [] + ).issubset({"process"}), + "post_runner_tool_schema_unchanged": post_runner_tool_surface.get("final_surface_matches_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 diff --git a/tests/test_leo_oos_readonly_guard.py b/tests/test_leo_oos_readonly_guard.py index 3265226..e7e7a4b 100644 --- a/tests/test_leo_oos_readonly_guard.py +++ b/tests/test_leo_oos_readonly_guard.py @@ -337,3 +337,32 @@ def test_registry_install_exposes_exact_schema_and_denies_ablation( assert surface["shell_tool_enabled"] is False assert len(surface["structured_read_tool_definition_sha256"]) == 64 assert denied["error_category"] == "kb_reads_disabled" + + registry._tools["process"] = FakeEntry() + post_runner = guard.reassert_read_only_tool_surface( + profile, + expected_kb_tool_sha256=source_sha256, + expected_definition_sha256=surface["structured_read_tool_definition_sha256"], + ) + + assert post_runner["initial_registry_tools"] == ["process", "skill_view", "skills_list", "teleo-kb"] + assert post_runner["removed_known_runner_tools"] == ["process"] + assert post_runner["actual_registry_tools"] == ["skill_view", "skills_list", "teleo-kb"] + assert post_runner["final_surface_matches_preexecution"] is True + + +def test_post_runner_registry_rejects_unknown_additions(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + profile, _kb_tool, source_sha256 = make_kb_tool(tmp_path) + registry = FakeRegistry() + registry._tools["teleo-kb"] = FakeEntry(schema=guard.STRUCTURED_READ_TOOL_SCHEMA) + registry._tools["unknown-network-tool"] = FakeEntry() + registry_module = types.ModuleType("tools.registry") + registry_module.registry = registry + monkeypatch.setitem(sys.modules, "tools.registry", registry_module) + + with pytest.raises(guard.ReadOnlyGuardError, match="unsupported additions"): + guard.reassert_read_only_tool_surface( + profile, + expected_kb_tool_sha256=source_sha256, + expected_definition_sha256="0" * 64, + ) diff --git a/tests/test_run_leo_m3taversal_oos_handler_suite.py b/tests/test_run_leo_m3taversal_oos_handler_suite.py index 2cf1400..848cc34 100644 --- a/tests/test_run_leo_m3taversal_oos_handler_suite.py +++ b/tests/test_run_leo_m3taversal_oos_handler_suite.py @@ -53,7 +53,9 @@ def test_guarded_remote_script_compiles_and_installs_preexecution_gate(grounding 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 "reassert_read_only_tool_surface" in script + assert "removed_known_runner_tools" in script + assert "final_surface_matches_preexecution" 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 diff --git a/tests/test_working_leo_m3taversal_oos_protocol.py b/tests/test_working_leo_m3taversal_oos_protocol.py index 7bbc248..dc8ae32 100644 --- a/tests/test_working_leo_m3taversal_oos_protocol.py +++ b/tests/test_working_leo_m3taversal_oos_protocol.py @@ -469,9 +469,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": { + "initial_registry_tools": ["process", "skill_view", "skills_list", "teleo-kb"], + "removed_known_runner_tools": ["process"], "actual_registry_tools": ["skill_view", "skills_list", "teleo-kb"], "structured_read_tool_definition_sha256": "a" * 64, - "unchanged_from_preexecution": True, + "final_surface_matches_preexecution": True, }, "readonly_guard_source_sha256": protocol["source_hashes"]["readonly_guard_sha256"], "safety_gate": {