test(runtime_wedge): module-functions signature snapshot drift gate

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) <noreply@anthropic.com>
This commit is contained in:
Hongming Wang 2026-04-30 07:01:10 -07:00
parent 4d1156cb8b
commit 70176e6c8f
3 changed files with 175 additions and 0 deletions

View File

@ -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: <name>, functions: [<sorted records>]}``
"""
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=...)

View File

@ -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"
}

View File

@ -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."
)