symfonic.services.shadow.context

context

What the body of a shadow run is handed — and nothing else.

The context is the whole surface a replacement implementation gets inside a shadow run: a gateway that cannot perform an effect, a state writer that cannot reach the store, and one way to call an extension. Extension calls are where the suppression claim is most easily lost, so every detection path records the defect on the context before it raises: a body with a bare except must not be able to buy back a claim the sentinel refused.

ShadowContext

ShadowContext(*, gateway: ShadowEffectGateway, state: SuppressedStateWriter, trust: ExtensionTrustRegistry, watch: bool)

What the body of a shadow run is given. Nothing else is reachable.

Source code in src/symfonic/services/shadow/context.py

def __init__(
    self,
    *,
    gateway: ShadowEffectGateway,
    state: SuppressedStateWriter,
    trust: ExtensionTrustRegistry,
    watch: bool,
) -> None:
    self.gateway = gateway
    self.state = state
    self._trust = trust
    self._watch = watch
    self._executed: list[str] = []
    self._defects: list[str] = []

claim_defects property

claim_defects: tuple[str, ...]

Reasons the suppression claim is void, independent of exceptions.

Detection must not depend on the exception reaching the harness: a body with a bare except would otherwise buy back the claim the sentinel just refused. Every detection path records here first and raises second.

call_extension

call_extension(extension_id: str, fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any

Run a synchronous extension only if its trust class permits it.

Opaque extensions raise before fn is touched. Port-mediated ones run inside the sentinel, so a wrong declaration becomes a MisdeclaredExtensionError instead of a silent effect. An extension that returns an awaitable is refused: its body would run after the probes came down. Use :meth:call_extension_async for those.

Source code in src/symfonic/services/shadow/context.py

def call_extension(
    self, extension_id: str, fn: Callable[..., Any], *args: Any, **kwargs: Any
) -> Any:
    """Run a synchronous extension only if its trust class permits it.

    Opaque extensions raise *before* ``fn`` is touched. Port-mediated ones
    run inside the sentinel, so a wrong declaration becomes a
    ``MisdeclaredExtensionError`` instead of a silent effect. An extension
    that returns an awaitable is refused: its body would run after the
    probes came down. Use :meth:`call_extension_async` for those.
    """
    self._trust.require_shadowable(extension_id)
    if not self._watch:
        self._executed.append(extension_id)
        return self._refuse_awaitable(extension_id, fn(*args, **kwargs))
    sentinel = DirectEffectSentinel(subject=extension_id, block=True)
    try:
        with sentinel.watching():
            result = fn(*args, **kwargs)
    except MisdeclaredExtensionError:
        self._witness(extension_id, sentinel)
        raise
    self._assert_clean(extension_id, sentinel)
    result = self._refuse_awaitable(extension_id, result)
    self._executed.append(extension_id)
    return result

call_extension_async async

call_extension_async(extension_id: str, fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any

Await an async extension inside the sentinel scope.

In an async-first framework the common tool shape is a coroutine function; calling one only builds the coroutine, so a synchronous call_extension would uninstall the probes before a single line of the extension ran. This awaits under the probes, which is the whole point of watching.

Source code in src/symfonic/services/shadow/context.py

async def call_extension_async(
    self, extension_id: str, fn: Callable[..., Any], *args: Any, **kwargs: Any
) -> Any:
    """Await an async extension *inside* the sentinel scope.

    In an async-first framework the common tool shape is a coroutine
    function; calling one only builds the coroutine, so a synchronous
    ``call_extension`` would uninstall the probes before a single line of
    the extension ran. This awaits under the probes, which is the whole
    point of watching.
    """
    self._trust.require_shadowable(extension_id)
    if not self._watch:
        self._executed.append(extension_id)
        return await _resolve(fn(*args, **kwargs))
    sentinel = DirectEffectSentinel(subject=extension_id, block=True)
    try:
        with sentinel.watching():
            result = await _resolve(fn(*args, **kwargs))
    except MisdeclaredExtensionError:
        self._witness(extension_id, sentinel)
        raise
    self._assert_clean(extension_id, sentinel)
    self._executed.append(extension_id)
    return result