symfonic.services.shadow.harness

harness

The effect-suppressed shadow harness.

One object owns the whole claim "this run performed no externally visible effect": it refuses to start against an incomplete classification, aborts before executing an opaque extension, routes every port call through the gateway, keeps replacement state writes in an overlay, and watches port-mediated extensions for direct effects.

run does not raise on abort. An abort is an outcome of a shadow run — the caller wants the ledger and the reason, not a traceback — so the result carries the status and raise_for_status re-raises on demand.

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

ShadowHarness

ShadowHarness(*, trust: ExtensionTrustRegistry, classification: EffectPortClassification = DEFAULT_EFFECT_CLASSIFICATION)

Builds and supervises shadow runs.

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

def __init__(
    self,
    *,
    trust: ExtensionTrustRegistry,
    classification: EffectPortClassification = DEFAULT_EFFECT_CLASSIFICATION,
) -> None:
    classification.assert_exhaustive()
    self._classification = classification
    self._trust = trust

ShadowRunResult dataclass

ShadowRunResult(run_id: str, tenant_id: str, status: ShadowStatus, ledger: EffectLedger, state_intents: tuple[StateWriteIntent, ...], extensions_executed: tuple[str, ...], body_executed: bool, suppression_claim: bool, claim_defects: tuple[str, ...] = (), abort_reason: str | None = None, error: BaseException | None = None, value: Any = None)

The reviewable artifact of one shadow run.

ShadowRunSpec dataclass

ShadowRunSpec(run_id: str, tenant_id: str, extensions: tuple[str, ...] = (), baseline_state: Mapping[str, Any] = dict(), backing_state: Mapping[str, Any] | None = None, responder: StubResponder | None = None, watch_extensions: bool = True)

What a shadow run is allowed to be before it starts.