From 70176e6c8f841fb836a8340260e33025d94d9eb6 Mon Sep 17 00:00:00 2001 From: Hongming Wang Date: Thu, 30 Apr 2026 07:01:10 -0700 Subject: [PATCH] test(runtime_wedge): module-functions signature snapshot drift gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BaseAdapter docstring tells adapter authors: > ``runtime_wedge.mark_wedged()`` / ``clear_wedge()`` — flip the > workspace to ``degraded`` + auto-recover when your SDK hits a > non-recoverable error class. Import directly from ``runtime_wedge``; > the heartbeat forwards the state to the platform automatically. That's a contract — adapter templates depend on the four module-level functions (``is_wedged``, ``wedge_reason``, ``mark_wedged``, ``clear_wedge``) being importable by those exact names with those exact signatures. Renaming any silently breaks every adapter that calls them: the import resolves the module fine, the ``AttributeError`` only surfaces when the adapter actually hits its first SDK error — long after the rename merges. Same drift class as #2378 / #2380 / #2381 (BaseAdapter, skill_loader) applied to the module-level function surface. Changes: - tests/_signature_snapshot.py gains build_module_functions_record. Walks a module's public top-level functions, optionally filtered to a specific name list (used here — runtime_wedge has internal helpers like reset_for_test that intentionally aren't part of the contract). Skips re-exports via __module__ check so a `from foo import bar` doesn't pollute the snapshot. - tests/test_runtime_wedge_signature.py snapshots the four contract functions. Plus a defense-in-depth required-functions test that catches removal even when source + snapshot are updated together. Verified: deliberately renaming `mark_wedged` → `mark_wedged_RENAMED` trips the gate with full snapshot diff in the failure message. Co-Authored-By: Claude Opus 4.7 (1M context) --- workspace/tests/_signature_snapshot.py | 41 ++++++++ .../snapshots/runtime_wedge_signature.json | 40 ++++++++ .../tests/test_runtime_wedge_signature.py | 94 +++++++++++++++++++ 3 files changed, 175 insertions(+) create mode 100644 workspace/tests/snapshots/runtime_wedge_signature.json create mode 100644 workspace/tests/test_runtime_wedge_signature.py diff --git a/workspace/tests/_signature_snapshot.py b/workspace/tests/_signature_snapshot.py index 709229c2..e6259007 100644 --- a/workspace/tests/_signature_snapshot.py +++ b/workspace/tests/_signature_snapshot.py @@ -93,6 +93,47 @@ def build_class_signature_record(cls: type) -> dict: return {"class": cls.__name__, "methods": methods} +def build_module_functions_record(module: object, function_names: list[str] | None = None) -> dict: + """Snapshot a module's public top-level functions. By default, walks + every public callable defined IN the module (excludes re-exports + via __module__ check). Pass ``function_names`` explicitly to pin a + specific set when the module exports more than the contract surface + (e.g. internal helpers that intentionally aren't part of the gate). + + Returns: ``{module: , functions: []}`` + """ + import types + + fns: list[dict] = [] + target_module = module.__name__ + + if function_names is not None: + for fn_name in sorted(function_names): + fn = getattr(module, fn_name, None) + if fn is None or not isinstance(fn, types.FunctionType): + # Caller asked for a name that isn't a function in the + # module — surface it as part of the snapshot so the + # error path stays in the failure-message-with-diff + # path rather than blowing up here. + fns.append({"name": fn_name, "missing": True}) + continue + fns.append(_signature_record(fn_name, fn)) + else: + for attr_name in sorted(vars(module)): + if attr_name.startswith("_"): + continue + attr = getattr(module, attr_name) + if not isinstance(attr, types.FunctionType): + continue + # Skip re-exports — only record functions defined IN this + # module so a `from foo import bar` doesn't pollute the + # snapshot. + if getattr(attr, "__module__", None) != target_module: + continue + fns.append(_signature_record(attr_name, attr)) + return {"module": target_module, "functions": fns} + + def build_dataclass_record(cls: type) -> dict: """Snapshot a dataclass's field shape. Captures field name + type annotation + has_default per field, plus the @dataclass(frozen=...) diff --git a/workspace/tests/snapshots/runtime_wedge_signature.json b/workspace/tests/snapshots/runtime_wedge_signature.json new file mode 100644 index 00000000..a4fec037 --- /dev/null +++ b/workspace/tests/snapshots/runtime_wedge_signature.json @@ -0,0 +1,40 @@ +{ + "functions": [ + { + "is_abstract": false, + "is_async": false, + "name": "clear_wedge", + "parameters": [], + "return_annotation": "None" + }, + { + "is_abstract": false, + "is_async": false, + "name": "is_wedged", + "parameters": [], + "return_annotation": "bool" + }, + { + "is_abstract": false, + "is_async": false, + "name": "mark_wedged", + "parameters": [ + { + "annotation": "str", + "has_default": false, + "kind": "POSITIONAL_OR_KEYWORD", + "name": "reason" + } + ], + "return_annotation": "None" + }, + { + "is_abstract": false, + "is_async": false, + "name": "wedge_reason", + "parameters": [], + "return_annotation": "str" + } + ], + "module": "runtime_wedge" +} diff --git a/workspace/tests/test_runtime_wedge_signature.py b/workspace/tests/test_runtime_wedge_signature.py new file mode 100644 index 00000000..0a345703 --- /dev/null +++ b/workspace/tests/test_runtime_wedge_signature.py @@ -0,0 +1,94 @@ +"""runtime_wedge public-API signature snapshot — drift gate. + +``BaseAdapter`` docstring explicitly tells adapter authors to call +``runtime_wedge.mark_wedged(reason)`` / ``clear_wedge()`` when their +SDK hits a non-recoverable error class — the heartbeat thread reads +``is_wedged()`` / ``wedge_reason()`` to flip the workspace to +``degraded`` and surface the cause to the canvas. + +That's a public adapter-facing API. Renaming any of the four +functions silently breaks every adapter that calls them: the import +still resolves the module, the missing attribute raises +``AttributeError`` only when the adapter actually hits its first +SDK error — long after the rename merges. + +Same drift class as the BaseAdapter signature snapshot (#2378, #2380) +and skill_loader gate (#2381), applied to the module-level +function surface. +""" + +import sys +from pathlib import Path + +import pytest + +WORKSPACE_DIR = Path(__file__).parent.parent +if str(WORKSPACE_DIR) not in sys.path: + sys.path.insert(0, str(WORKSPACE_DIR)) + +from tests._signature_snapshot import ( # noqa: E402 + build_module_functions_record, + compare_against_snapshot, +) + +SNAPSHOT_PATH = Path(__file__).parent / "snapshots" / "runtime_wedge_signature.json" + + +def _build_full_snapshot() -> dict: + """Pin only the four contract functions adapters call. Other module- + level helpers (``reset_for_test``, internal state) intentionally + aren't part of the snapshot — adapters MUST NOT depend on them. + """ + import runtime_wedge + + return build_module_functions_record( + runtime_wedge, + function_names=[ + "is_wedged", + "wedge_reason", + "mark_wedged", + "clear_wedge", + ], + ) + + +def test_runtime_wedge_signature_matches_snapshot(): + compare_against_snapshot(_build_full_snapshot(), SNAPSHOT_PATH) + + +def test_snapshot_has_required_functions(): + """Defense-in-depth: even if both source and snapshot are updated + together, removing any of the four adapter-facing functions + requires explicit edit here. The required set is the documented + public contract — see ``BaseAdapter`` docstring. + """ + if not SNAPSHOT_PATH.exists(): + pytest.skip(f"{SNAPSHOT_PATH.name} not generated yet") + + import json + snapshot = json.loads(SNAPSHOT_PATH.read_text()) + fn_names = {f["name"] for f in snapshot["functions"]} + + required = { + "is_wedged", # platform-side heartbeat reads this + "wedge_reason", # surfaces the why on the canvas + "mark_wedged", # adapters call this on non-recoverable errors + "clear_wedge", # adapters call this on auto-recovery + } + missing = required - fn_names + if missing: + pytest.fail( + f"runtime_wedge snapshot is missing required functions: {sorted(missing)}.\n" + "Either restore them on runtime_wedge.py, OR coordinate adapter " + "updates AND remove the entry from `required` in this test " + "with a justification." + ) + + for fn in snapshot["functions"]: + if fn.get("missing"): + pytest.fail( + f"runtime_wedge.{fn['name']} resolved as a non-function — " + "either it was replaced by a different kind of attribute " + "(class? module-level alias?) which adapters' direct call " + "would break, OR it was removed entirely." + )