molecule-core/workspace/tests/test_runtime_capabilities.py
Hongming Wang 205a454c09 feat(runtime): RuntimeCapabilities dataclass + BaseAdapter.capabilities()
Foundation primitive for the native+pluggable runtime principle (task
#117, blocks #87). Lets each adapter declare which cross-cutting
capabilities it owns natively (heartbeat, scheduler, durable session,
status mgmt, retry, activity decoration, channel dispatch) versus
delegates to the platform's fallback implementation.

Pure additive: every existing adapter inherits BaseAdapter.capabilities()
which returns RuntimeCapabilities() — every flag False — so today's
"platform owns everything" behavior is preserved exactly. Subsequent
PRs land platform-side consumers (idle-timeout override, scheduler
skip, status-transition hook, etc.) one capability at a time.

Why a frozen dataclass instead of class attributes: capabilities are
declared at class-load time and read by the platform on every heartbeat.
A mutable value would let a runtime change capabilities mid-flight,
creating impossible-to-debug state where the platform's idea of who-
owns-heartbeat drifts from the adapter's actual code.

Why a `to_dict()` with explicit short keys: the Go side will read these
from the heartbeat payload by string key. The dict's wire names are
pinned independently of Python field names so a Python-side rename
doesn't silently break the Go consumer (test pins this).

Tests (9 new):
  - is a frozen dataclass (mutation rejected)
  - all 7 default flags are False (load-bearing — flipping any default
    silently moves ownership for langgraph/crewai/deepagents)
  - to_dict() keys are stable wire names (Go contract)
  - BaseAdapter.capabilities() default returns all-False
  - subclass override mechanism works
  - sibling adapters' defaults aren't affected by an override

Verification:
  - 1300/1300 workspace pytest pass (was 1291, +9)
  - Zero behavior change for any existing code path

See project memory `project_runtime_native_pluggable.md`.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-26 22:17:49 -07:00

155 lines
6.1 KiB
Python

"""Tests for RuntimeCapabilities + BaseAdapter.capabilities() — the
foundation primitive for the native+pluggable runtime principle (task
#117). The dataclass + default method are intentionally a no-op
addition; these tests pin that contract so a future change can't
accidentally flip a default and silently move ownership.
"""
from dataclasses import is_dataclass
import pytest
from adapter_base import BaseAdapter, RuntimeCapabilities
class _MinimalAdapter(BaseAdapter):
"""Concrete subclass with only the abstract members satisfied —
every other behavior should fall through to BaseAdapter defaults
so we can assert what those defaults are."""
@staticmethod
def name() -> str:
return "test-minimal"
@staticmethod
def display_name() -> str:
return "Test Minimal"
@staticmethod
def description() -> str:
return "Minimal adapter for capability default tests"
async def setup(self, config) -> None:
return None
async def create_executor(self, config): # pragma: no cover
raise NotImplementedError
class _NativeHeartbeatAdapter(_MinimalAdapter):
"""Models a runtime that owns heartbeat natively — declares it via
capabilities() override. Used to verify the override mechanism
works without touching defaults."""
def capabilities(self) -> RuntimeCapabilities:
return RuntimeCapabilities(provides_native_heartbeat=True)
class TestRuntimeCapabilitiesDataclass:
"""The dataclass surface itself."""
def test_is_a_dataclass(self):
assert is_dataclass(RuntimeCapabilities)
def test_is_frozen(self):
# Immutability matters: capabilities are declared at class-load
# time and read by the platform on every heartbeat. A mutable
# value would let a runtime change capabilities mid-flight,
# creating impossible-to-debug state where the platform's idea
# of who-owns-heartbeat drifts from the adapter's actual code.
c = RuntimeCapabilities()
with pytest.raises((AttributeError, Exception)):
c.provides_native_heartbeat = True # type: ignore[misc]
def test_all_defaults_false(self):
# Every flag MUST default to False — that's what makes adding
# the dataclass a no-op for existing adapters. If any default
# flips to True, every adapter that didn't override capabilities
# silently switches who-owns-that-capability and the platform
# stops providing the fallback. Catastrophic for langgraph /
# crewai / deepagents which have no native impl.
c = RuntimeCapabilities()
assert c.provides_native_heartbeat is False
assert c.provides_native_scheduler is False
assert c.provides_native_session is False
assert c.provides_native_status_mgmt is False
assert c.provides_native_retry is False
assert c.provides_activity_decoration is False
assert c.provides_channel_dispatch is False
def test_to_dict_keys_are_stable_wire_names(self):
# The Go side reads these by string key from the heartbeat
# payload. If Python renames a field (provides_native_heartbeat
# → has_native_heartbeat) the dict's wire name should NOT change
# — pin the JSON keys here so a refactor on the Python side
# doesn't silently break the Go consumer.
c = RuntimeCapabilities()
assert set(c.to_dict().keys()) == {
"heartbeat",
"scheduler",
"session",
"status_mgmt",
"retry",
"activity_decoration",
"channel_dispatch",
}
def test_to_dict_values_match_flags(self):
c = RuntimeCapabilities(
provides_native_heartbeat=True,
provides_native_session=True,
)
d = c.to_dict()
assert d["heartbeat"] is True
assert d["session"] is True
# Untouched flags stay False — we don't want a "True for one
# capability flips siblings via dataclass inheritance" surprise.
assert d["scheduler"] is False
assert d["status_mgmt"] is False
class TestBaseAdapterCapabilitiesDefault:
"""The BaseAdapter.capabilities() default — the contract every
existing adapter inherits without changes."""
def test_default_returns_all_false(self):
# The whole point of landing this primitive as a separate PR
# is that it's behavior-preserving for everyone. If this test
# fails, every adapter in the project has just had its
# capability declarations silently changed.
a = _MinimalAdapter()
caps = a.capabilities()
assert caps == RuntimeCapabilities()
assert caps.to_dict() == {
"heartbeat": False,
"scheduler": False,
"session": False,
"status_mgmt": False,
"retry": False,
"activity_decoration": False,
"channel_dispatch": False,
}
def test_default_returns_RuntimeCapabilities_instance(self):
a = _MinimalAdapter()
assert isinstance(a.capabilities(), RuntimeCapabilities)
def test_subclass_can_override_capabilities(self):
# Without this working, the entire native+pluggable principle
# is unimplementable. Pin it with a fixture that flips one flag.
a = _NativeHeartbeatAdapter()
caps = a.capabilities()
assert caps.provides_native_heartbeat is True
# Sibling flags untouched — overriding one doesn't accidentally
# move ownership of the others.
assert caps.provides_native_scheduler is False
assert caps.provides_native_session is False
def test_override_does_not_affect_default_for_other_subclasses(self):
# Method-level dispatch, not class-attribute mutation. A
# subclass declaring native_heartbeat must NOT change what
# _MinimalAdapter (a sibling) reports.
minimal = _MinimalAdapter().capabilities()
native = _NativeHeartbeatAdapter().capabilities()
assert minimal.provides_native_heartbeat is False
assert native.provides_native_heartbeat is True