1d231ed295
Prerequisite for the universal-runtime refactor (task #87) to move claude_sdk_executor.py out of molecule-runtime into the claude-code template repo. heartbeat.py had a hard import: from claude_sdk_executor import is_wedged, wedge_reason which would break the moment the executor moves out of the runtime package — the heartbeat would lose access to the wedge state used to flip workspace status to degraded. Extract the wedge state to a runtime-side module that the heartbeat can keep importing regardless of which adapter executor is wedged: - workspace/runtime_wedge.py — single-flag state + mark_wedged / clear_wedge / is_wedged / wedge_reason / reset_for_test. Same semantics as the original claude_sdk_executor implementation (sticky first-write-wins, auto-clear on observed success). 100 LOC of pure stateless helpers; lock-free ok because there's one executor per workspace process today. - workspace/claude_sdk_executor.py — drops the in-file definitions; re-exports the same names from runtime_wedge as a backwards-compat shim. Any third-party adapter that imported is_wedged / wedge_reason / _mark_sdk_wedged from claude_sdk_executor keeps working for one release cycle while they migrate to runtime_wedge. - workspace/heartbeat.py — _runtime_state_payload() now imports from runtime_wedge instead of claude_sdk_executor. Lazy-import pattern preserved; the docstring updated to explain the new cross-cutting source-of-truth. Tests (10 new in test_runtime_wedge.py): - Default state (unwedged), mark sets flag, first-write-wins, clear restores healthy, clear-when-not-wedged is no-op, re-marking after clear is allowed - Re-export shim: each old name in claude_sdk_executor IS the runtime_wedge function (identity check), state is shared (marking via the executor shim is observable via runtime_wedge and vice versa) Verification: - 1251/1251 workspace pytest pass (was 1241 after orphan deletion; +10 = exactly the new test_runtime_wedge.py cases) - All existing test_claude_sdk_executor.py cases (which call _mark_sdk_wedged via the shim) still pass After this lands + the claude-code template image rebuilds with the local claude_sdk_executor.py copy (template PR #13), the molecule- core deletion of workspace/claude_sdk_executor.py becomes safe (the shim deletion comes alongside the file deletion, since runtime_wedge is the new public API). See project memory `project_runtime_native_pluggable.md`. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
104 lines
4.5 KiB
Python
104 lines
4.5 KiB
Python
"""Tests for runtime_wedge — the runtime-side wedge-state module that
|
|
heartbeat reads + adapter executors write. Extracted from claude_sdk_
|
|
executor (task #87 universal-runtime refactor) so the executor can move
|
|
to its template repo without breaking heartbeat.
|
|
|
|
The behavior is identical to the prior in-executor implementation; tests
|
|
pin the contract so the re-export shim in claude_sdk_executor.py can
|
|
later be deleted without surprise."""
|
|
import pytest
|
|
|
|
import runtime_wedge
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset():
|
|
"""Each test starts with a clean wedge state — production wedges are
|
|
sticky-per-process, but cross-test bleed would couple unrelated cases."""
|
|
runtime_wedge.reset_for_test()
|
|
yield
|
|
runtime_wedge.reset_for_test()
|
|
|
|
|
|
class TestRuntimeWedge:
|
|
def test_starts_unwedged(self):
|
|
assert runtime_wedge.is_wedged() is False
|
|
assert runtime_wedge.wedge_reason() == ""
|
|
|
|
def test_mark_wedged_sets_flag_and_reason(self):
|
|
runtime_wedge.mark_wedged("SDK init timeout")
|
|
assert runtime_wedge.is_wedged() is True
|
|
assert runtime_wedge.wedge_reason() == "SDK init timeout"
|
|
|
|
def test_first_mark_wins(self):
|
|
# Stable banner text is more important than the most-recent
|
|
# cause. A second wedge while already wedged should NOT
|
|
# overwrite — operator sees the original (more diagnosable)
|
|
# reason, not whatever the SDK said next.
|
|
runtime_wedge.mark_wedged("SDK init timeout")
|
|
runtime_wedge.mark_wedged("Subsequent identical-class wedge")
|
|
assert runtime_wedge.wedge_reason() == "SDK init timeout"
|
|
|
|
def test_clear_wedge_restores_healthy(self):
|
|
# Auto-recovery: when the SDK starts working again, the next
|
|
# heartbeat must report empty runtime_state so the platform
|
|
# flips status from degraded back to online.
|
|
runtime_wedge.mark_wedged("transient blip")
|
|
runtime_wedge.clear_wedge()
|
|
assert runtime_wedge.is_wedged() is False
|
|
assert runtime_wedge.wedge_reason() == ""
|
|
|
|
def test_clear_wedge_when_not_wedged_is_noop(self):
|
|
# No-op safety — production calls clear_wedge() on every
|
|
# successful query (~thousands of times per session); throwing
|
|
# or logging when not wedged would spam.
|
|
runtime_wedge.clear_wedge()
|
|
runtime_wedge.clear_wedge() # still safe twice in a row
|
|
assert runtime_wedge.is_wedged() is False
|
|
|
|
def test_re_marking_after_clear_is_allowed(self):
|
|
# Real production path: SDK wedges, recovers, wedges again.
|
|
# Each cycle should land cleanly (not silently drop).
|
|
runtime_wedge.mark_wedged("first wedge")
|
|
runtime_wedge.clear_wedge()
|
|
runtime_wedge.mark_wedged("second wedge — different reason")
|
|
assert runtime_wedge.is_wedged() is True
|
|
assert runtime_wedge.wedge_reason() == "second wedge — different reason"
|
|
|
|
|
|
class TestClaudeSdkExecutorReExportShim:
|
|
"""claude_sdk_executor.py keeps re-exporting the old names for one
|
|
release cycle so any third-party adapter copying our wedge convention
|
|
has time to migrate. These tests pin the shim — when removed, the
|
|
test file goes too."""
|
|
|
|
def test_is_wedged_re_exported(self):
|
|
from claude_sdk_executor import is_wedged
|
|
assert is_wedged is runtime_wedge.is_wedged
|
|
|
|
def test_wedge_reason_re_exported(self):
|
|
from claude_sdk_executor import wedge_reason
|
|
assert wedge_reason is runtime_wedge.wedge_reason
|
|
|
|
def test_internal_helpers_re_exported(self):
|
|
# Keep the underscore names too — claude_sdk_executor's own
|
|
# _run_query calls _mark_sdk_wedged / _clear_sdk_wedge_on_success
|
|
# via these re-exports.
|
|
from claude_sdk_executor import (
|
|
_mark_sdk_wedged,
|
|
_clear_sdk_wedge_on_success,
|
|
_reset_sdk_wedge_for_test,
|
|
)
|
|
assert _mark_sdk_wedged is runtime_wedge.mark_wedged
|
|
assert _clear_sdk_wedge_on_success is runtime_wedge.clear_wedge
|
|
assert _reset_sdk_wedge_for_test is runtime_wedge.reset_for_test
|
|
|
|
def test_re_export_state_is_shared(self):
|
|
# The shim isn't a copy — both names refer to the same module
|
|
# state. Marking via the executor name must be observable via
|
|
# the runtime_wedge name (and vice versa).
|
|
from claude_sdk_executor import _mark_sdk_wedged
|
|
_mark_sdk_wedged("via executor shim")
|
|
assert runtime_wedge.is_wedged() is True
|
|
assert runtime_wedge.wedge_reason() == "via executor shim"
|