"""ABI-1 PluginAPI 2.0 negotiation matrix (v7next Ф3.1-B). Pins the full manifest-negotiation contract: absent field ≡ the LEGACY generation by construction ("1.3", deliberately NOT 1.4); declared versions are held to major-strict / minor-as-minimum; capabilities form a closed set validated against the actual execution mode; every refusal is typed and educational; and the per-version surface fingerprint fails closed in BOTH directions (unknown version, and live-surface drift without a version bump). """ from __future__ import annotations import pytest from ouroboros.contracts import plugin_api as papi from ouroboros.contracts.plugin_api import ( ExecutionMode, LEGACY_PLUGIN_API_GENERATION, PLUGIN_API_SURFACE_FINGERPRINTS, PLUGIN_API_VERSION, api_generation, extension_new_pass_admission_error, negotiate_plugin_api, plugin_api_surface_fingerprint, ) from ouroboros.contracts.skill_manifest import ( SkillManifestError, parse_skill_manifest_text, ) from tests._extension_loader_shared import _prepare_extension from tests._extension_loader_shared import ( # noqa: F401 (autouse fixture applies on import) _clear_loader_state, ) def _manifest(plugin_api_line: str = "") -> object: return parse_skill_manifest_text( "---\n" "name: matrix_ext\n" "description: d\n" "version: 1.0.0\n" "type: extension\n" "entry: plugin.py\n" "permissions: [tool]\n" f"{plugin_api_line}" "---\n" "body\n" ) # --- parse shape (fail-closed structure, tolerant absence) ------------------- def test_plugin_api_field_is_optional_and_absent_means_legacy_generation(): manifest = _manifest() assert manifest.plugin_api is None assert api_generation(manifest) == LEGACY_PLUGIN_API_GENERATION == "1.3" result = negotiate_plugin_api(manifest) assert result.ok and not result.declared assert result.generation == "1.3" def test_plugin_api_field_parses_string_and_mapping_forms(): as_string = _manifest('plugin_api: "2.0"\n') assert as_string.plugin_api == {"version": "2.0", "capabilities": []} as_mapping = _manifest( "plugin_api:\n version: \"2.0\"\n capabilities: [register_tool, log]\n" ) assert as_mapping.plugin_api == { "version": "2.0", "capabilities": ["register_tool", "log"], } @pytest.mark.parametrize("bad_line", [ "plugin_api: [2, 0]\n", "plugin_api:\n version: \"2.0\"\n extra_key: true\n", "plugin_api:\n capabilities: [register_tool]\n", ]) def test_plugin_api_structural_damage_fails_manifest_parse(bad_line): with pytest.raises(SkillManifestError): _manifest(bad_line) def test_invalid_declaration_surfaces_as_manifest_validate_warning(): manifest = _manifest('plugin_api: "1.0"\n') warnings = manifest.validate() assert any("plugin_api" in w for w in warnings) # --- negotiation: major strict / minor minimum / typed education ------------- def test_exact_host_version_negotiates_to_the_host_generation(): result = negotiate_plugin_api(_manifest('plugin_api: "2.0"\n')) assert result.ok and result.declared assert result.generation == PLUGIN_API_VERSION def test_lower_major_is_refused_with_grandfather_education(): result = negotiate_plugin_api(_manifest('plugin_api: "1.3"\n')) assert not result.ok assert "major is strict" in result.error assert "OMIT the field" in result.error def test_higher_major_is_refused(): result = negotiate_plugin_api(_manifest('plugin_api: "3.0"\n')) assert not result.ok and "major" in result.error def test_higher_minor_minimum_is_refused_with_upgrade_education(): result = negotiate_plugin_api(_manifest('plugin_api: "2.7"\n')) assert not result.ok assert "minimum minor" in result.error and "upgrade" in result.error def test_malformed_version_string_is_refused_typed(): result = negotiate_plugin_api(_manifest('plugin_api: "two.zero"\n')) assert not result.ok and "major.minor" in result.error # --- capabilities: closed set + execution-mode availability ------------------ def test_unknown_capability_is_refused_and_names_the_closed_set(): result = negotiate_plugin_api(_manifest( "plugin_api:\n version: \"2.0\"\n capabilities: [register_rootkit]\n" )) assert not result.ok assert "register_rootkit" in result.error and "closed" in result.error def test_mode_unavailable_capability_is_refused_with_companion_education(): manifest = _manifest( "plugin_api:\n version: \"2.0\"\n capabilities: [subscribe_event]\n" ) in_proc = negotiate_plugin_api(manifest, mode=ExecutionMode.IN_PROCESS) assert in_proc.ok and in_proc.capabilities == ("subscribe_event",) out_proc = negotiate_plugin_api(manifest, mode=ExecutionMode.OUT_OF_PROCESS) assert not out_proc.ok and "companion_process" in out_proc.error # --- surface fingerprint: fail-closed in both directions --------------------- def test_recorded_fingerprint_matches_the_live_surface(): assert PLUGIN_API_SURFACE_FINGERPRINTS[PLUGIN_API_VERSION] == plugin_api_surface_fingerprint() def test_unknown_version_key_fails_closed(monkeypatch): monkeypatch.setattr( papi, "PLUGIN_API_SURFACE_FINGERPRINTS", {PLUGIN_API_VERSION: plugin_api_surface_fingerprint(), "2.9": "unused"}, ) monkeypatch.setattr(papi, "PLUGIN_API_VERSION", "2.9") # Declared "2.0" is inside major/minor range for host 2.9 but has no # recorded fingerprint entry in this simulated table -> refused. monkeypatch.setitem(papi.PLUGIN_API_SURFACE_FINGERPRINTS, "2.9", papi.plugin_api_surface_fingerprint()) del papi.PLUGIN_API_SURFACE_FINGERPRINTS["2.0"] result = papi.negotiate_plugin_api(_manifest('plugin_api: "2.0"\n')) assert not result.ok and "no recorded surface fingerprint" in result.error def test_live_surface_drift_without_version_bump_fails_closed(monkeypatch): monkeypatch.setitem( papi.PLUGIN_API_SURFACE_FINGERPRINTS, PLUGIN_API_VERSION, "0" * 64, ) drifted = papi.negotiate_plugin_api(_manifest('plugin_api: "2.0"\n')) assert not drifted.ok and "drifted" in drifted.error # The drift refusal is host-integrity: even the grandfather path refuses. legacy = papi.negotiate_plugin_api(_manifest()) assert not legacy.ok and "drifted" in legacy.error # --- admission predicate shape (full issuance-path coverage lives in # tests/test_plugin_api_admission.py) ---------------------------------------- def test_admission_error_only_for_fieldless_or_invalid_extensions(): assert extension_new_pass_admission_error(None) == "" fieldless = _manifest() assert "plugin_api" in extension_new_pass_admission_error(fieldless) declared = _manifest('plugin_api: "2.0"\n') assert extension_new_pass_admission_error(declared) == "" invalid = _manifest('plugin_api: "1.0"\n') assert "major is strict" in extension_new_pass_admission_error(invalid) instruction = parse_skill_manifest_text("# just_instructions\nbody\n") assert extension_new_pass_admission_error(instruction) == "" # --- loader integration: negotiation gates the load, generation rides # the published surfaces ------------------------------------------------------ def test_declared_2_0_extension_loads_and_carries_the_generation(tmp_path): from ouroboros import extension_loader from ouroboros.extension_surface_names import extension_surface_name loaded, _repo, drive_root = _prepare_extension( tmp_path, "gen2ext", plugin_body=( "def register(api):\n" " api.register_tool('t1', lambda **kw: 'ok', description='d', schema={})\n" ), permissions=["tool"], extra_frontmatter='plugin_api: "2.0"\n', ) err = extension_loader.load_extension(loaded, lambda: {}, drive_root=drive_root, _force_in_process=True) assert err is None, err entry = extension_loader.get_tool(extension_surface_name("gen2ext", "t1")) assert entry is not None and entry.get("plugin_api_generation") == "2.0" with extension_loader._lock: bundle = extension_loader._extensions["gen2ext"] assert bundle.plugin_api_generation == "2.0" def test_grandfathered_fieldless_extension_still_loads_as_legacy(tmp_path): from ouroboros import extension_loader from ouroboros.extension_surface_names import extension_surface_name loaded, _repo, drive_root = _prepare_extension( tmp_path, "legacyext", plugin_body=( "def register(api):\n" " api.register_tool('t1', lambda **kw: 'ok', description='d', schema={})\n" ), permissions=["tool"], ) err = extension_loader.load_extension(loaded, lambda: {}, drive_root=drive_root, _force_in_process=True) assert err is None, err entry = extension_loader.get_tool(extension_surface_name("legacyext", "t1")) assert entry is not None assert entry.get("plugin_api_generation") == LEGACY_PLUGIN_API_GENERATION def test_refused_negotiation_blocks_the_load_before_import(tmp_path): from ouroboros import extension_loader loaded, _repo, drive_root = _prepare_extension( tmp_path, "badgen", plugin_body="RAN = True\ndef register(api):\n pass\n", permissions=["tool"], extra_frontmatter='plugin_api: "9.9"\n', ) err = extension_loader.load_extension(loaded, lambda: {}, drive_root=drive_root, _force_in_process=True) assert err is not None and "negotiation refused" in err import sys from ouroboros.extension_import_staging import _module_key assert _module_key("badgen") not in sys.modules, ( "plugin.py was imported although negotiation refused the load" ) # --- surface fingerprint: SHAPE, not just names ------------------------------ def _mutant_module(tmp_path, old: str, new: str): """Load a copy of the contract with one surface edit applied. The copy keeps the RECORDED fingerprint table verbatim — that is the whole point: a shape change without a version bump must make the live surface disagree with what the release recorded, so negotiation refuses. """ import importlib.util import pathlib import sys source = pathlib.Path(papi.__file__).read_text(encoding="utf-8") assert source.count(old) == 1, old target = tmp_path / "plugin_api_mutant.py" target.write_text(source.replace(old, new), encoding="utf-8") spec = importlib.util.spec_from_file_location("plugin_api_mutant", target) module = importlib.util.module_from_spec(spec) # @dataclass resolves its own module through sys.modules while the class # body executes, so the entry must exist before exec_module. sys.modules[spec.name] = module try: spec.loader.exec_module(module) finally: sys.modules.pop(spec.name, None) return module @pytest.mark.parametrize(("old", "new", "what"), ( ("def get_settings(self, keys: Sequence[str]) -> Dict[str, Any]:", "def get_settings(self, keys: List[str]) -> Dict[str, Any]:", "a method's parameter annotation"), (" server_port: int\n", " server_port: str\n", "a RuntimeInfo value type"), ), ids=("method-parameter-annotation", "runtime_info-value-type")) def test_a_shape_change_without_a_version_bump_fails_closed(tmp_path, old, new, what): """The digest hashed method NAMES and RuntimeInfo KEYS only, so every signature-level break — the exact class this module's header says requires a version bump — left the recorded digest byte-identical and negotiated happily. Both directions of the fingerprint contract were therefore blind to the changes an extension author actually collides with.""" mutant = _mutant_module(tmp_path, old, new) assert mutant.plugin_api_surface_fingerprint() != plugin_api_surface_fingerprint(), what refused = mutant.negotiate_plugin_api(_manifest('plugin_api: "2.0"\n')) assert not refused.ok and "drifted" in refused.error, what def test_the_fingerprint_covers_every_public_method_signature_and_runtime_key(): """Naming the payload's two halves so a future narrowing is visible: the digest must move for ANY public method's signature text and for ANY RuntimeInfo annotation, not merely for an added or removed name.""" import inspect methods = sorted( m for m in dir(papi.PluginAPI) if not m.startswith("_") and callable(getattr(papi.PluginAPI, m, None)) ) assert methods, "the surface cannot be empty" for name in methods: assert str(inspect.signature(getattr(papi.PluginAPI, name))) assert set(papi.RuntimeInfo.__annotations__) >= {"server_port", "capabilities"}