"""Private Dify decisions cross native ordinary-error conversion without I/O. This opt-in component does not authorize, persist, dispatch, or issue a waiting receipt. Trusted composition must bind its instance-owned context and sink. """ from collections.abc import Generator from contextlib import AbstractContextManager, contextmanager from dataclasses import dataclass from datetime import datetime from functools import cache from sys import exc_info from typing import Any, ClassVar, Literal, Protocol, runtime_checkable from uuid import uuid4 from pydantic import ValidationError from graphon.engine_events.base import NodeEvent from graphon.engine_events.node import NodeRunPauseRequestedEvent from graphon.entities.pause_reason import SchedulingPause from graphon.nodes.base.node import Node from graphon.nodes.container_effects import ContainerAwaitRequest, ContainerRunResult from runtime_contracts.builder.native.contracts import ( ApprovalWait, DispatchPermit, EffectDecision, NativeTestContextRef, PolicyStop, RecordedResult, WaitReason, ) class _NativeControl(BaseException): """Only this private family bypasses native ordinary Exception handlers.""" def __init__(self, *, ref: NativeTestContextRef, diagnostic: str) -> None: super().__init__(diagnostic) self.ref = ref class NativeApprovalWait(_NativeControl): """Host-owned wait category and pending operation, never message authority.""" def __init__(self, *, ref: NativeTestContextRef, pending_operation_id: str, wait_reason: WaitReason) -> None: super().__init__(ref=ref, diagnostic="native_control_wait") self.pending_operation_id = pending_operation_id self.wait_reason = wait_reason class NativePolicyStop(_NativeControl): """Host-owned fixed reason, carried separately from safe diagnostics.""" def __init__(self, *, ref: NativeTestContextRef, reason_code: str) -> None: super().__init__(ref=ref, diagnostic="native_control_stop") self.reason_code = reason_code class NativeControlBridge: """Convert only validated decisions; successful values keep their identity.""" @staticmethod def resolve(*, ref: NativeTestContextRef, decision: EffectDecision) -> DispatchPermit | RecordedResult: if type(decision) not in (DispatchPermit, RecordedResult, ApprovalWait, PolicyStop): raise NativePolicyStop(ref=ref, reason_code="native_control_invalid_decision") try: # Frozen DTOs can still be constructed/copied without validation. type(decision).model_validate(decision.model_dump(warnings=False), strict=True) except (ValidationError, ValueError, TypeError): raise NativePolicyStop(ref=ref, reason_code="native_control_invalid_decision") from None if isinstance(decision, DispatchPermit | RecordedResult): return decision if isinstance(decision, ApprovalWait): raise NativeApprovalWait( ref=ref, pending_operation_id=decision.pending_operation_id, wait_reason=decision.wait_reason ) raise NativePolicyStop(ref=ref, reason_code=decision.reason_code) @dataclass(frozen=True, slots=True) class NativeControlRecord: """Private typed facts; carrier_ref correlates only and grants no authority.""" control: NativeApprovalWait | NativePolicyStop node_id: str native_execution_id: str carrier_ref: str @dataclass(frozen=True, slots=True) class NativeControlAccepted: """Sink recorded this exact object and sealed stop authority if needed.""" record: NativeControlRecord @dataclass(frozen=True, slots=True) class NativeControlRefused: """Sink refusal contains no arbitrary diagnostics or resumable authority.""" @runtime_checkable class NativeControlSink(Protocol): """Trusted synchronous host port, supplied explicitly by composition. Validate the complete owner/context chain, retain the exact typed control, and for stop seal actual dispatch authority before returning acceptance of the same record. Refusal or any callback exception becomes UNKNOWN. A successful return is not a checkpoint commit or a public waiting receipt. """ def record_control(self, *, record: NativeControlRecord) -> NativeControlAccepted | NativeControlRefused: ... BoundaryStatus = Literal["OPEN", "CONTROL_PENDING", "WAIT_RECORDED", "STOP_RECORDED", "UNKNOWN_NONRESUMABLE"] class NativeNodeOwnership(Protocol): """Observe actual bound node work, including synchronous control callbacks. Composition retains the supplied node/runtime and current thread through scope exit. This port grants no dispatch, checkpoint or wait authority. """ def node_scope(self, *, node: Node, phase: Literal["run", "resume_container"]) -> AbstractContextManager[None]: ... class BoundaryOwnerState: """Instance-owned in-memory facts, shared by all child frames of this owner. This is NOT a durable fence, generation claim, checkpoint eligibility check, capacity owner or physical cancellation mechanism. Trusted composition must consult dispatch_admissible before every new dispatch and retain/project UNKNOWN or STOP independently of native pause/success events. Never reopen this object on resume; separately admitted composition supplies a new owner. Omitting ownership preserves isolated-component compatibility only. Future host qualification requires its explicit observer and separate settlement. """ def __init__(self, *, ref: NativeTestContextRef, ownership: NativeNodeOwnership | None = None) -> None: if type(ref) is not NativeTestContextRef: raise ValueError("native_control_invalid_ref") NativeTestContextRef.model_validate(ref.model_dump(warnings=False), strict=True) self.ref = ref self._ownership = ownership self._records: list[NativeControlRecord] = [] self._accepted: set[str] = set() self._reason_code: str | None = None @property def records(self) -> tuple[NativeControlRecord, ...]: return tuple(self._records) @property def reason_code(self) -> str | None: return self._reason_code @property def dispatch_admissible(self) -> bool: return not self._records and self._reason_code is None @property def status(self) -> BoundaryStatus: if self._reason_code is not None: return "UNKNOWN_NONRESUMABLE" if any( isinstance(record.control, NativePolicyStop) and record.carrier_ref in self._accepted for record in self._records ): return "STOP_RECORDED" if len(self._accepted) != len(self._records): return "CONTROL_PENDING" return "WAIT_RECORDED" if self._records else "OPEN" def _begin(self, record: NativeControlRecord) -> None: # Retain facts and deny new dispatch before calling host code. Each # record moves pending -> accepted, or owner-wide sticky UNKNOWN. self._records.append(record) def _unknown(self, reason_code: str) -> None: if self._reason_code is None: self._reason_code = reason_code NativeEventStream = Generator[NodeEvent | ContainerAwaitRequest] class _NativeControlBoundary: """Mixin selected only by marked composition, never by registry replacement.""" _native_control_node_class: ClassVar[type[Node]] def __init__( self, *args: Any, native_control_owner: BoundaryOwnerState, native_control_sink: NativeControlSink, **kwargs: Any, ) -> None: if ( not isinstance(self, Node) or type(native_control_owner) is not BoundaryOwnerState or not isinstance(native_control_sink, NativeControlSink) or not callable(native_control_sink.record_control) ): raise ValueError("native_control_owner_and_sink_required") self._native_control_owner = native_control_owner self._native_control_sink = native_control_sink self._native_control_node = self self._native_control_node_class.__init__(self._native_control_node, *args, **kwargs) def _abort_control(self, reason_code: str) -> None: """Contain only public abort-call failure; never relabel node failures.""" try: self._native_control_node.runtime_state.graph_execution.abort(reason_code) except BaseException: # No callback/host failure may escape a native worker. Native # settlement is unqualified if this public abort itself fails. self._native_control_owner._unknown("native_control_abort_unavailable") def _control_execution_id(self) -> str | None: try: execution_id = self._native_control_node.execution_id except RuntimeError: execution_id = "" if not execution_id: self._native_control_owner._unknown("native_control_execution_id_missing") self._abort_control("native_control_execution_id_missing") return None return execution_id def _record_control(self, control: NativeApprovalWait | NativePolicyStop, execution_id: str) -> NativeEventStream: node = self._native_control_node owner = self._native_control_owner record = NativeControlRecord( control=control, node_id=node.id, native_execution_id=execution_id, carrier_ref=uuid4().hex ) owner._begin(record) if control.ref != owner.ref: owner._unknown("native_control_ref_mismatch") self._abort_control("native_control_ref_mismatch") return try: response = self._native_control_sink.record_control(record=record) except BaseException: # Intentionally confined to this explicit host callback. Foreign # BaseException from the native delegate is never caught here. owner._unknown("native_control_sink_failed") self._abort_control("native_control_sink_failed") return if type(response) is not NativeControlAccepted or response.record is not record: owner._unknown("native_control_sink_refused") self._abort_control("native_control_sink_refused") return owner._accepted.add(record.carrier_ref) if owner.status in ("UNKNOWN_NONRESUMABLE", "STOP_RECORDED"): self._abort_control("native_control_nonresumable") return yield NodeRunPauseRequestedEvent( id=execution_id, node_id=node.id, node_type=node.node_type, reason=SchedulingPause(message=record.carrier_ref), ) @contextmanager def _observe_node(self, phase: Literal["run", "resume_container"]) -> Generator[bool]: """Contain observer failures only, ignoring any exception suppression. Entry is lazy and follows execution binding validation. The scope stays active across yielded events and typed control/sink conversion. Exit is finally-safe; an exit failure makes any earlier pause nonresumable while a native exception (including GeneratorExit) keeps its exact identity. """ ownership = self._native_control_owner._ownership if ownership is None: yield True return try: scope = ownership.node_scope(node=self._native_control_node, phase=phase) scope.__enter__() except BaseException: self._native_control_owner._unknown("native_control_observation_failed") self._abort_control("native_control_observation_failed") yield False return try: yield True finally: try: # Explicit calls keep observer suppression/replacement from # changing native control or foreign exception semantics. scope.__exit__(*exc_info()) except BaseException: self._native_control_owner._unknown("native_control_observation_failed") self._abort_control("native_control_observation_failed") def run(self) -> NativeEventStream: execution_id = self._control_execution_id() if execution_id is None: return with self._observe_node("run") as entered: if not entered: return try: yield from self._native_control_node_class.run(self._native_control_node) except (NativeApprovalWait, NativePolicyStop) as control: yield from self._record_control(control, execution_id) def resume_container(self, *, result: ContainerRunResult, started_at: datetime) -> NativeEventStream: execution_id = self._control_execution_id() if execution_id is None: return with self._observe_node("resume_container") as entered: if not entered: return try: yield from self._native_control_node_class.resume_container( self._native_control_node, result=result, started_at=started_at ) except (NativeApprovalWait, NativePolicyStop) as control: yield from self._record_control(control, execution_id) @cache def _adapt_concrete_node(node_class: type[Node]) -> type[Node]: # Cache concrete classes only: no host/ref/sink/runtime captured in closure. # Native subclass registration increments the public registry version, so # downstream mapping caches must honor Graphon's version invalidation. adapted = type( f"NativeControl{node_class.__name__}", (_NativeControlBoundary, node_class), {"__module__": __name__, "_native_control_node_class": node_class}, ) assert issubclass(adapted, Node) return adapted def marked_native_node_class(node_class: type[Node], *, marked: bool) -> type[Node]: """Adapt an already-resolved concrete class only on an explicit True marker. The real factory must finish native/extension loading and resolve its current type/version first, including after registry version changes. This cache is keyed by concrete class identity, never by a stale type mapping. Existing native type/version/latest entries are left untouched. Constructors require native_control_owner and native_control_sink as instance arguments. """ if type(marked) is not bool: raise ValueError("native_control_concrete_class_required") if not marked: return node_class if not issubclass(node_class, Node) or issubclass(node_class, _NativeControlBoundary): raise ValueError("native_control_concrete_class_required") versions = Node.get_node_type_classes_mapping().get(node_class.node_type, {}) if str(node_class.version()) not in versions: raise ValueError("native_control_native_class_not_loaded") return _adapt_concrete_node(node_class)