molecule-ai/molecule-core
35
0
Fork 2
Code Issues 16 Pull Requests 2 Actions 49 Packages Projects Releases Wiki Activity

refactor(wedge): address review feedback — class wrap + import-path doc + dedupe shim rationale

Browse Source 
Three changes from /code-review-and-quality on PR #2154:

1. Optional (architecture): wrap state in a private _WedgeState class
   instead of bare module-level globals. Public API (mark_wedged /
   clear_wedge / is_wedged / wedge_reason / reset_for_test) is
   unchanged — adapters never see the class. The class is forward-cover
   for any future per-scope variant (multiple executors per process, a
   keyed registry, etc.) without churning the call sites. Today there's
   exactly one instance (_DEFAULT) so behavior is identical.

2. Optional (readability): clarify the import path in the integration
   recipe — in a TEMPLATE repo it's `from molecule_runtime.runtime_wedge`
   (PyPI package); in molecule-core itself it's `from runtime_wedge`
   (top-level module). Removes the trap where a contributor reading the
   docstring while editing in-repo copies the template-style import and
   gets ImportError.

3. Nit (readability): dedupe the shim rationale. claude_sdk_executor's
   re-export comment now points to runtime_wedge's "Compatibility shim"
   section as the source of truth instead of restating the same content.
   Avoids docs-in-two-places drift risk.

Verification:
  - 1251/1251 workspace pytest pass (no behavior change — class wrap
    is pure plumbing; module-level helpers delegate to the singleton)
  - All shim re-export identity tests still pass (the shim's
    `is_wedged is runtime_wedge.is_wedged` assertion holds because we
    re-export the SAME function object that delegates to _DEFAULT)

No new tests needed — the existing test suite covers the public API
contract; the class is an implementation detail behind that contract.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Hongming Wang 2026-04-27 00:16:33 -07:00
parent cd899c969f
commit feb544938b
2 changed files with 70 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
workspace/claude_sdk_executor.py
View File

@ -87,19 +87,11 @@ _RETRYABLE_PATTERNS = (
"try again",
)
# SDK-wedge state lives in the runtime-side module (runtime_wedge) so
# heartbeat.py and any future cross-cutting consumer can read it without
# importing this adapter-specific executor. Decoupling was the prerequisite
# for moving claude_sdk_executor out of molecule-runtime into the
# claude-code template repo (task #87 — universal-runtime refactor).
#
# Local re-exports keep the in-file call sites (_run_query etc.) terse
# and preserve the historical names so the behavior is identical to
# the pre-extraction version. is_wedged/wedge_reason are also re-exported
# so any external consumer that imported them from this module keeps
# working — heartbeat.py has been updated to import from runtime_wedge
# directly, but a third-party adapter copying our wedge convention may
# still expect them here.
# Wedge state moved to runtime_wedge (see that module's docstring for
# the rationale + the broader "Compatibility shim" note). This block
# re-exports under the historical names so the in-file call sites in
# _run_query stay terse and any external consumer that imported them
# from claude_sdk_executor keeps working for one release cycle.
from runtime_wedge import ( # noqa: E402
clear_wedge as _clear_sdk_wedge_on_success,
is_wedged,

84
workspace/runtime_wedge.py
View File

@ -40,6 +40,13 @@ in place — adapters do not change anything in molecule-runtime.
Minimum integration (~6 LOC inside the executor):
# Import path:
# - In a TEMPLATE repo (the common case for new adapters), the
# runtime is installed via PyPI as `molecule-ai-workspace-runtime`,
# so the import is `from molecule_runtime.runtime_wedge import …`.
# - In molecule-core itself (when editing this repo's own
# workspace/ tree), the module is at the top level — import as
# `from runtime_wedge import …`.
from molecule_runtime.runtime_wedge import mark_wedged, clear_wedge
async def execute(self, ctx, queue):
@ -95,23 +102,72 @@ import logging
logger = logging.getLogger(__name__)
# Single-flag state. None = healthy; non-empty string = wedged with that
# human-readable reason. Surfaced verbatim as the canvas's degraded-card
# banner text via heartbeat.sample_error.
_wedged_reason: str | None = None
class _WedgeState:
"""Internal carrier for the wedge flag. Exposed only via the module-
level helpers below; adapters never see this class.
Wrapping the state in a class (instead of a bare module-level global)
is forward-cover for the day a runtime hosts multiple executors per
process — a future per-scope variant can hand out keyed instances
without changing the public mark_wedged / clear_wedge / is_wedged /
wedge_reason API. Today there's exactly one instance (_DEFAULT).
"""
def __init__(self) -> None:
# None = healthy; non-empty string = wedged with that human-
# readable reason. Surfaced verbatim as the canvas's degraded-
# card banner text via heartbeat.sample_error.
self._reason: str | None = None
def is_wedged(self) -> bool:
return self._reason is not None
def reason(self) -> str:
return self._reason or ""
def mark(self, reason: str) -> None:
# First-write-wins: a subsequent identical-class wedge can't
# overwrite a more specific initial reason so the operator-
# visible banner stays stable.
if self._reason is None:
self._reason = reason
logger.error(
"runtime wedge detected: %s — workspace will report degraded until cleared",
reason,
)
def clear(self) -> None:
# No-op when not wedged (the common case — adapters call this
# on every successful query).
if self._reason is not None:
logger.info(
"runtime wedge cleared after successful operation — workspace will recover to online on next heartbeat",
)
self._reason = None
def reset(self) -> None:
# Unconditional clear — for test fixtures only. Skips the
# info-level log line the production clear() path emits.
self._reason = None
# Single shared instance backing the module-level helpers. Today there's
# one executor per workspace process so this fits perfectly; the class
# wrap above is the seam for any future per-scope variant.
_DEFAULT = _WedgeState()
def is_wedged() -> bool:
"""True if some adapter executor in this process has marked itself
wedged. Sticky until the same executor calls clear_wedge() on
observed recovery (or the process restarts)."""
return _wedged_reason is not None
return _DEFAULT.is_wedged()
def wedge_reason() -> str:
"""Human-readable description of the wedge cause, or empty string
when not wedged. Surfaced to the canvas via heartbeat sample_error."""
return _wedged_reason or ""
return _DEFAULT.reason()
def mark_wedged(reason: str) -> None:
@ -123,13 +179,7 @@ def mark_wedged(reason: str) -> None:
SDK has hit a non-recoverable error class. Safe to call multiple
times; the no-op when already wedged is intentional.
"""
global _wedged_reason
if _wedged_reason is None:
_wedged_reason = reason
logger.error(
"runtime wedge detected: %s — workspace will report degraded until cleared",
reason,
)
_DEFAULT.mark(reason)
def clear_wedge() -> None:
@ -142,10 +192,7 @@ def clear_wedge() -> None:
and the platform flips status back to online.
No-op when not wedged (the common case)."""
global _wedged_reason
if _wedged_reason is not None:
logger.info("runtime wedge cleared after successful operation — workspace will recover to online on next heartbeat")
_wedged_reason = None
_DEFAULT.clear()
def reset_for_test() -> None:
@ -153,5 +200,4 @@ def reset_for_test() -> None:
clear_wedge() on observed success; this helper is for unit tests
that need to reset between cases without going through the full
SDK round-trip."""
global _wedged_reason
_wedged_reason = None
_DEFAULT.reset()