dify/api/services/dify_builder/native_launch_observation.py

"""Observe the native worker boundary, without granting execution authority.

Instrumented generation and failed root preparation can prove execution never
began. Missing instrumentation and interrupted processes stay unknown.
"""

import re
from collections.abc import Generator
from contextlib import contextmanager
from contextvars import ContextVar
from dataclasses import dataclass, field
from typing import Literal

FailureResource = tuple[Literal["model", "tool"], str, str]


@dataclass
class NativeLaunchObservation:
    preparing: bool = False
    dispatched: bool = False
    failure_resource: FailureResource | None = None
    failure_node_id: str | None = None
    _resource_node_observed: bool = field(default=False, repr=False)
    _resource_error: Exception | None = field(default=None, repr=False)
    initialization_error: Exception | None = field(default=None, repr=False)
    _initialization_claimed: bool = field(default=False, repr=False)

    def resource_for(self, error: Exception) -> FailureResource | None:
        current: BaseException | None = error
        for _ in range(8):
            if current is None:
                break
            if current is self._resource_error:
                return self.failure_resource
            current = current.__cause__ or current.__context__
        return None

    def node_ids_for(self, error: Exception, graph: dict) -> tuple[str, ...]:
        resource = self.resource_for(error)
        if resource is None:
            return ()
        if self._resource_node_observed:
            return (self.failure_node_id,) if self.failure_node_id in resource_node_ids(graph, resource) else ()
        return resource_node_ids(graph, resource)

    @contextmanager
    def track(self) -> Generator[None]:
        token = _active.set(self)
        try:
            yield
        finally:
            _active.reset(token)

    @property
    def failed_before_dispatch(self) -> bool:
        return self.preparing and not self.dispatched


_active: ContextVar[NativeLaunchObservation | None] = ContextVar("builder_native_launch", default=None)


def begin_preparation() -> None:
    if (observation := _active.get()) is not None:
        observation.preparing = True


def mark_dispatched() -> None:
    from services.dify_builder.task_controls import check_current

    check_current()
    # Mark before Thread.start: a start acknowledgement failure is conservative.
    if (observation := _active.get()) is not None:
        observation.dispatched = True


@contextmanager
def observe_initialization() -> Generator[None]:
    """Observe only the first root preparation of this dispatched launch.

    The orchestrator calls this before any engine can run. A successful first
    preparation permanently closes the boundary, so nested workflow failures
    cannot claim that the parent did not execute. Only an escaping, attributed
    metadata exception is retained; no stream text grants this proof.
    """
    observation = _active.get()
    if observation is None or not observation.dispatched or observation._initialization_claimed:
        yield
        return
    observation._initialization_claimed = True
    try:
        yield
    except Exception as exc:
        if observation.resource_for(exc) is not None:
            observation.initialization_error = exc
        raise


def _public_id(value: str) -> bool:
    return "://" not in value and bool(re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_./:@+-]{0,127}", value))


@contextmanager
def observe_resource(
    kind: Literal["model", "tool"], provider: str, name: str = "", *, node_id: object = None
) -> Generator[None]:
    """Attribute only a failed metadata read, never successful earlier reads."""
    try:
        yield
    except Exception as exc:
        observation = _active.get()
        if observation is not None:
            from services.dify_builder import trace_diagnostics

            if trace_diagnostics.plugin_activation_unavailable(exc):
                # Worker context is copied before dispatch. Observe the typed
                # cause here, before native response handling converts it to
                # text; this grants no execution or retry authority.
                trace_diagnostics.report(
                    "native_preparation_failed" if observation.failed_before_dispatch else "native_outcome_unknown",
                    error=exc,
                    provider=provider,
                    model=name if kind == "model" else None,
                )
        if observation is not None and _public_id(provider) and (not name or _public_id(name)):
            observation.failure_resource = (kind, provider, name)
            observation._resource_node_observed = node_id is not None
            observation.failure_node_id = node_id if isinstance(node_id, str) and _public_id(node_id) else None
            observation._resource_error = exc
        raise


def resource_node_ids(graph: dict, resource: FailureResource | None) -> tuple[str, ...]:
    if resource is None:
        return ()
    kind, provider, name = resource
    result = []
    for node in graph.get("nodes", []):
        data = node.get("data") or {}
        if kind == "model":
            config = data.get("model") or {}
            matches = data.get("type") == "llm" and config.get("provider") == provider and config.get("name") == name
        else:
            matches = (
                data.get("type") == "tool"
                and data.get("provider_id") == provider
                and (not name or data.get("tool_name") == name)
            )
        node_id = node.get("id")
        if matches and isinstance(node_id, str) and _public_id(node_id):
            result.append(node_id)
    return tuple(sorted(set(result)))