Add reporting, retry lifecycle, and dry-run functionality

ccflow/evaluators/__init__.py

@@ -1,2 +1,4 @@
 from .common import *
+from .dry_run import *
+from .reporting import *
 from .retry import *

ccflow/evaluators/common.py

@@ -1,13 +1,8 @@
-import itertools
 import logging
-import time
-from contextlib import nullcontext
-from datetime import timedelta
-from pprint import pformat
 from types import MappingProxyType
 from typing import Any, Callable, Dict, List, Optional, Set, Union
 
-from pydantic import Field, PrivateAttr, field_validator
+from pydantic import Field, PrivateAttr
 from typing_extensions import override
 
 from ..base import BaseModel, make_lazy_result
@@ -21,12 +16,15 @@
     TransparentModelEvaluationContext,
     WrapperModel,
 )
+from ..utils.reporting import FormatConfig, LoggingPolicy
 from ..utils.tokenize import compute_cache_token
+from .reporting import ReportingEvaluator, _descriptor
 
 __all__ = [
     "cache_key",
     "combine_evaluators",
     "FallbackEvaluator",
+    "FormatConfig",
     "LazyEvaluator",
     "LoggingEvaluator",
     "MemoryCacheEvaluator",
@@ -158,106 +156,23 @@ def make_result():
         return make_lazy_result(context.model.result_type, make_result)
 
 
-class FormatConfig(BaseModel):
-    """Configuration for formatting the result of the evaluation.
-
-    This is used by the LoggingEvaluator to control how the result is formatted.
-    """
-
-    arrow_as_polars: bool = Field(
-        False,
-        description="Whether to convert pyarrow tables to polars tables for formatting, as arrow formatting does not work well with large tables or provide control over options",
-    )
-    pformat_config: Dict[str, Any] = Field({}, description="pformat config to use for formatting data")
-    polars_config: Dict[str, Any] = Field({}, description="polars config to use for formatting polars frames")
-    pandas_config: Dict[str, Any] = Field({}, description="pandas config to use for formatting pandas objects")
-
-
-class LoggingEvaluator(EvaluatorBase):
+class LoggingEvaluator(ReportingEvaluator, LoggingPolicy):
     """Evaluator that logs information about evaluating the callable.
 
-    It logs start and end times, the model name, and the context."""
-
-    log_level: int = Field(logging.DEBUG, description="The log level for start/end of evaluation")
-    verbose: bool = Field(True, description="Whether to output the model definition as part of logging")
-    log_result: bool = Field(False, description="Whether to log the result of the evaluation")
-    format_config: FormatConfig = Field(FormatConfig(), description="Configuration for formatting the result of the evaluation if log_result=True")
-
-    def is_transparent(self, context: ModelEvaluationContext) -> bool:
-        return True
-
-    @field_validator("log_level", mode="before")
-    @classmethod
-    def _validate_log_level(cls, v: Union[int, str]) -> int:
-        """Validate that the log level is a valid logging level."""
-        if isinstance(v, str):
-            return getattr(logging, v.upper(), "")
-        return v
+    It logs start and end times, the model name, and the context. This is the *default* evaluator
+    when no other is configured. It is now a thin combination of :class:`ReportingEvaluator` (span /
+    contextvar correlation, optional structured events when a ``reporter`` is set) and
+    :class:`~ccflow.utils.reporting.LoggingPolicy` (the actual log output, preserved exactly), so it
+    also participates in the reporting span tree.
+    """
 
     @override
     def __call__(self, context: ModelEvaluationContext) -> ResultType:
-        model_name = context.model.meta.name or context.model.__class__.__name__
-        log_level = context.options.get("log_level", self.log_level)
-        verbose = context.options.get("verbose", self.verbose)
-        log.log(log_level, "[%s]: Start evaluation of %s on %s.", model_name, context.fn, context.context)
-        if verbose:
-            log.log(log_level, "[%s]: %s", model_name, context.model)
-        start = time.time()
-        result = None
-        try:
-            result = context()
-            return result
-        finally:
-            end = time.time()
-            if self.log_result and result is not None:
-                log.log(
-                    log_level,
-                    self._format_result(result),
-                    model_name,
-                    context.fn,
-                    context.context,
-                )
-            log.log(
-                log_level,
-                "[%s]: End evaluation of %s on %s (time elapsed: %s).",
-                model_name,
-                context.fn,
-                context.context,
-                timedelta(seconds=end - start),
-            )
-
-    def _format_result(self, result: ResultType) -> str:
-        """Handle formatting of the result"""
-        # Add special formatting for eager table/data frame types embedded in the results
-        import pyarrow as pa
-
-        result_dict = result.model_dump(by_alias=True)
-        for k, v in result_dict.items():
-            try:
-                if self.format_config.arrow_as_polars and isinstance(v, pa.Table):
-                    import polars as pl  # Only import polars if needed
-
-                    result_dict[k] = pl.from_arrow(v)
-            except TypeError:
-                pass
-
-        if self.format_config.polars_config:  # Control formatting of polars tables if set
-            import polars as pl  # Only import polars if needed
-
-            polars_context = pl.Config(**self.format_config.polars_config)
-        else:
-            polars_context = nullcontext()
-
-        if self.format_config.pandas_config:  # Control formatting of pandas tables if set
-            import pandas as pd
-
-            pandas_context = pd.option_context(*itertools.chain.from_iterable(self.format_config.pandas_config.items()))
-        else:
-            pandas_context = nullcontext()
-
-        with polars_context, pandas_context:
-            msg_str = "[%s]: Result of %s on %s:\n"
-            return f"{msg_str}{pformat(result_dict, **self.format_config.pformat_config)}"
+        return self._run_with_reporting(
+            context,
+            extra={"model": context.model, "raw_context": context.context, "options": dict(context.options)},
+            **_descriptor(context),
+        )
 
 
 def _effective_model_key(
@@ -466,7 +381,12 @@ def __deepcopy__(self, memo):
 
 
 class CallableModelGraph(BaseModel):
-    """Class to hold a "graph" """
+    """Dependency graph of callable-model evaluation contexts.
+
+    ``graph`` maps each node's cache key to the set of its dependency cache keys, ``ids`` maps cache
+    keys back to their :class:`ModelEvaluationContext`, and ``root_id`` is the cache key of the node
+    the graph was built from.
+    """
 
     graph: Dict[bytes, Set[bytes]]
     ids: Dict[bytes, ModelEvaluationContext]
@@ -552,7 +472,7 @@ def __call__(self, context: ModelEvaluationContext) -> ResultType:
         import graphlib
 
         # If we are evaluating deps, or if we have already started using the graph evaluator further up the call tree,
-        # no not apply it any further
+        # do not apply it any further
         if self._is_evaluating:
             return context()
         self._is_evaluating = True

ccflow/evaluators/dry_run.py

@@ -0,0 +1,86 @@
+"""Dependency-graph planning without model execution."""
+
+from collections import deque
+from contextvars import ContextVar
+
+from pydantic import Field
+from typing_extensions import override
+
+from ..callable import EvaluatorBase, ModelEvaluationContext, ResultType
+from ..utils.reporting import ReportContext, ReportingPolicy, ReportPhase, _new_span_id
+from .reporting import _model_type
+
+__all__ = ["DryRunEvaluator"]
+
+
+_DRY_RUN_PLANNING: ContextVar[bool] = ContextVar("ccflow_dry_run_planning", default=False)
+
+
+class DryRunEvaluator(EvaluatorBase):
+    """Plan a dependency graph without running model bodies.
+
+    Dependency declarations are evaluated to discover the graph. When ``reporting`` has a reporter,
+    each node emits ``QUEUED`` and ``SKIPPED`` events. The default synthetic result is constructed
+    without validation and must not be used in downstream computation.
+    """
+
+    reporting: ReportingPolicy = Field(default_factory=ReportingPolicy)
+    synthetic_result: bool = Field(
+        True,
+        description="If True, return an unvalidated synthetic result. If False, run the wrapped evaluation after planning.",
+    )
+
+    def is_transparent(self, context: ModelEvaluationContext) -> bool:
+        return not self.synthetic_result
+
+    @override
+    def __call__(self, context: ModelEvaluationContext) -> ResultType:
+        if _DRY_RUN_PLANNING.get() or context.fn == "__deps__":
+            return context()
+
+        from .common import _flatten_cache_key_context, cache_key, get_dependency_graph
+
+        token = _DRY_RUN_PLANNING.set(True)
+        try:
+            graph = get_dependency_graph(context)
+            if self.reporting.reporter is not None:
+                span_ids = {key: _new_span_id() for key in graph.graph}
+                parent_of: dict = {}
+                depth_of: dict = {graph.root_id: 0}
+                order: list = []
+                seen: set = set()
+                queue = deque([graph.root_id])
+                while queue:
+                    key = queue.popleft()
+                    if key in seen:
+                        continue
+                    seen.add(key)
+                    order.append(key)
+                    for child in graph.graph.get(key, ()):
+                        parent_of.setdefault(child, key)
+                        depth_of.setdefault(child, depth_of.get(key, 0) + 1)
+                        queue.append(child)
+
+                for key in order:
+                    evaluation_context = graph.ids[key]
+                    flattened, fn, _ = _flatten_cache_key_context(evaluation_context)
+                    model = flattened.model
+                    logical_key = cache_key(ModelEvaluationContext(model=model, context=flattened.context, fn=fn, options=flattened.options))
+                    report_context = ReportContext(
+                        model_name=model.meta.name or model.__class__.__name__,
+                        model_type=_model_type(model),
+                        fn=fn,
+                        context_repr=self.reporting._context_repr(flattened.context),
+                        span_id=span_ids[key],
+                        parent_span_id=span_ids.get(parent_of.get(key)),
+                        depth=depth_of.get(key, 0),
+                        extra={"node_key": logical_key.decode("utf-8"), "dry_run": True},
+                    )
+                    self.reporting._emit(self.reporting._event(report_context, ReportPhase.QUEUED, extra=report_context.extra))
+                    self.reporting._emit(self.reporting._event(report_context, ReportPhase.SKIPPED, extra=report_context.extra))
+
+            if self.synthetic_result:
+                return context.model.result_type.model_construct()
+            return context()
+        finally:
+            _DRY_RUN_PLANNING.reset(token)

ccflow/evaluators/reporting.py

@@ -0,0 +1,111 @@
+"""Reporting (telemetry / tracing / metrics / alerting) evaluators.
+
+These are the cross-cutting ("how to run") way to attach reporting to a callable graph: a
+:class:`ReportingEvaluator` wraps the evaluation of every model in its scope and is configured at
+runtime via ``FlowOptions`` / ``FlowOptionsOverride``. The reporting mechanics come from the shared
+policies in :mod:`ccflow.utils.reporting`, mirroring how ``RetryEvaluator`` reuses ``RetryPolicy``.
+
+A reporting evaluator is always *transparent*: a successful evaluation returns exactly the same value
+as evaluating the wrapped context directly, so it does not affect cache keys or dependency-graph
+deduplication. Reporting describes the *evaluation* (model identity, context, timing, graph topology,
+failures) -- it never acts on the result payload, which is what distinguishes it from a publisher.
+
+The class taxonomy is ``<Vendor><Signal>ReportingEvaluator`` where ``Signal`` is one of ``Tracing``,
+``Metrics`` or ``Alerts``. OpenTelemetry tracing and metrics are implemented.
+"""
+
+from typing_extensions import override
+
+from ..callable import EvaluatorBase, ModelEvaluationContext, ResultType
+from ..exttypes import PyObjectPath
+from ..utils.reporting import (
+    AlertsPolicy,
+    MetricsPolicy,
+    OpenTelemetryMetricsPolicy,
+    OpenTelemetryTracingPolicy,
+    ReportingPolicy,
+    TracingPolicy,
+)
+
+__all__ = [
+    "ReportingEvaluator",
+    "TracingReportingEvaluator",
+    "MetricsReportingEvaluator",
+    "AlertsReportingEvaluator",
+    "OpenTelemetryTracingReportingEvaluator",
+    "OpenTelemetryMetricsReportingEvaluator",
+    "OpenTelemetryEvaluator",
+]
+
+
+def _model_type(model) -> str:
+    """Best-effort fully-qualified type name for a model.
+
+    Falls back to ``module.qualname`` when the type cannot be resolved as an importable
+    :class:`PyObjectPath` (e.g. dynamically-generated classes such as Ray-wrapped callables), so that
+    reporting never breaks evaluation.
+    """
+    try:
+        return str(PyObjectPath.validate(type(model)))
+    except Exception:
+        cls = type(model)
+        return f"{cls.__module__}.{cls.__qualname__}"
+
+
+def _descriptor(context: ModelEvaluationContext) -> dict:
+    """Build the reporting descriptor (model name/type/fn/context) for an evaluation context."""
+    model = context.model
+    return {
+        "model_name": model.meta.name or model.__class__.__name__,
+        "model_type": _model_type(model),
+        "fn": context.fn,
+        "context": context.context,
+    }
+
+
+class ReportingEvaluator(EvaluatorBase, ReportingPolicy):
+    """Evaluator that reports telemetry about the evaluation of the callable models in its scope.
+
+    This is the base of the reporting evaluator family. It is *transparent* (the result is identical
+    to evaluating the wrapped context directly) and keeps no mutable per-call state on the instance --
+    all per-call state lives in local variables / context vars -- so a single instance can be shared
+    safely across threads and combined with parallel evaluators (Ray / Celery).
+
+    Configure a :class:`~ccflow.utils.reporting.Reporter` to receive the emitted
+    :class:`~ccflow.utils.reporting.ReportEvent` objects. To attach reporting to a single specific
+    model as part of the graph itself, use :class:`~ccflow.models.reporting.ReportingModel` instead.
+    """
+
+    def is_transparent(self, context: ModelEvaluationContext) -> bool:
+        return True
+
+    @override
+    def __call__(self, context: ModelEvaluationContext) -> ResultType:
+        return self._run_with_reporting(context, **_descriptor(context))
+
+
+class TracingReportingEvaluator(ReportingEvaluator, TracingPolicy):
+    """Reporting evaluator specialised for distributed *tracing* (spans)."""
+
+
+class MetricsReportingEvaluator(ReportingEvaluator, MetricsPolicy):
+    """Reporting evaluator specialised for *metrics* (counters / latency histograms)."""
+
+
+class AlertsReportingEvaluator(ReportingEvaluator, AlertsPolicy):
+    """Reporting evaluator specialised for *alerting*, with ``P1``-``P5`` priority tags."""
+
+
+class OpenTelemetryTracingReportingEvaluator(TracingReportingEvaluator, OpenTelemetryTracingPolicy):
+    """Tracing evaluator backed by OpenTelemetry spans (requires ``opentelemetry-api``)."""
+
+
+class OpenTelemetryMetricsReportingEvaluator(MetricsReportingEvaluator, OpenTelemetryMetricsPolicy):
+    """Metrics evaluator backed by OpenTelemetry counters/histograms (requires ``opentelemetry-api``)."""
+
+
+OpenTelemetryEvaluator = OpenTelemetryTracingReportingEvaluator
+
+# Future reporting backends may include Datadog tracing, metrics, and alerts; Opsgenie metrics and
+# alerts; Jira Service Management alerts; and New Relic tracing, metrics, and alerts. Add each public
+# evaluator only when its backend policy is implemented.

ccflow/models/__init__.py

@@ -1,3 +1,4 @@
 from .narwhals import *
 from .publisher import *
+from .reporting import *
 from .retry import *