diff --git a/lib/crewai/src/crewai/flow/expressions.py b/lib/crewai/src/crewai/flow/expressions.py index 2250124df..63c79566e 100644 --- a/lib/crewai/src/crewai/flow/expressions.py +++ b/lib/crewai/src/crewai/flow/expressions.py @@ -2,7 +2,9 @@ from __future__ import annotations -from collections.abc import Iterable +from collections.abc import Callable, Iterable +from dataclasses import dataclass +from datetime import datetime, timezone from functools import lru_cache import json from typing import TYPE_CHECKING, Any, NamedTuple, TypeAlias, cast @@ -21,6 +23,46 @@ _CEL_MACROS_WITH_LOCAL_BINDINGS = frozenset( ) +@dataclass(frozen=True) +class _CelRunContext: + now: datetime + + +@dataclass(frozen=True) +class _CelFunctionSpec: + annotation: Any + factory: Callable[[_CelRunContext], Callable[..., Any]] + + +@lru_cache(maxsize=1) +def _cel_function_registry() -> dict[str, _CelFunctionSpec]: + from celpy import celtypes + + return { + "now": _CelFunctionSpec( + annotation=celtypes.FunctionType, + factory=lambda run: lambda: celtypes.TimestampType(run.now), + ), + } + + +def _cel_environment() -> Any: + from celpy import Environment + + return Environment( + annotations={ + name: spec.annotation for name, spec in _cel_function_registry().items() + } + ) + + +def _cel_functions(run_context: _CelRunContext) -> dict[str, Any]: + return { + name: spec.factory(run_context) + for name, spec in _cel_function_registry().items() + } + + def _find_cel_eval_error(value: Any) -> Exception | None: from celpy.evaluation import CELEvalError @@ -103,6 +145,9 @@ FLOW_TEMPLATE_EXPRESSION_RULES: tuple[str, ...] = ( "Use this for numbers, booleans, objects, and lists.", "If the string has other text, the final value is text. Non-text values " "become JSON. `null` becomes empty text.", + "Use `now()` for the current UTC time as a CEL timestamp, frozen for the " + "whole run. Use standard CEL on it: `string(now())` for ISO text, " + "`now().getFullYear()`, or `now() - duration('24h')`.", ) FLOW_TEMPLATE_EXPRESSION_CONTRACT = " ".join(FLOW_TEMPLATE_EXPRESSION_RULES) FLOW_TEMPLATE_EXPRESSION_EXAMPLES: dict[str, tuple[dict[str, str], ...]] = { @@ -176,10 +221,15 @@ class Expression: """CEL expression helper used for definition-time checks and runtime rendering.""" def __init__( - self, value: ExpressionData, *, context: dict[str, Any] | None = None + self, + value: ExpressionData, + *, + context: dict[str, Any] | None = None, + now: datetime | None = None, ) -> None: self.value = value self.context = context + self.now = now @classmethod def from_flow( @@ -190,7 +240,11 @@ class Expression: local_context: dict[str, Any] | None = None, ) -> Expression: """Build an expression with the standard Flow runtime context.""" - return cls(value, context=cls._flow_context(flow, local_context=local_context)) + return cls( + value, + context=cls._flow_context(flow, local_context=local_context), + now=getattr(flow, "_cel_now", None), + ) def validate_expression( self, @@ -231,6 +285,7 @@ class Expression: return self._evaluate_cel( self._require_cel_source(cast(str, self.value)), resolved_context or {}, + self._run_context(), ) def render_template(self, context: dict[str, Any] | None = None) -> Any: @@ -240,7 +295,12 @@ class Expression: type; strings mixing literals and expressions render as text. """ resolved_context = self.context if context is None else context - return self._render_template_value(self.value, resolved_context or {}) + return self._render_template_value( + self.value, resolved_context or {}, self._run_context() + ) + + def _run_context(self) -> _CelRunContext: + return _CelRunContext(now=self.now or datetime.now(timezone.utc)) @staticmethod def _validate_template_value( @@ -311,20 +371,27 @@ class Expression: return context @staticmethod - def _render_template_value(value: ExpressionData, context: dict[str, Any]) -> Any: + def _render_template_value( + value: ExpressionData, context: dict[str, Any], run_context: _CelRunContext + ) -> Any: if isinstance(value, str): - return Expression._render_template_string(value, context) + return Expression._render_template_string(value, context, run_context) if isinstance(value, dict): return { - key: Expression._render_template_value(item, context) + key: Expression._render_template_value(item, context, run_context) for key, item in value.items() } if isinstance(value, list): - return [Expression._render_template_value(item, context) for item in value] + return [ + Expression._render_template_value(item, context, run_context) + for item in value + ] return value @staticmethod - def _render_template_string(value: str, context: dict[str, Any]) -> Any: + def _render_template_string( + value: str, context: dict[str, Any], run_context: _CelRunContext + ) -> Any: segments = _parse_template_segments(value) expressions = [ segment for segment in segments if isinstance(segment, _ExpressionSegment) @@ -333,26 +400,28 @@ class Expression: return value literals = [segment for segment in segments if isinstance(segment, str)] if len(expressions) == 1 and all(not literal.strip() for literal in literals): - return Expression._evaluate_cel(expressions[0].source, context) + return Expression._evaluate_cel(expressions[0].source, context, run_context) rendered: list[str] = [] for segment in segments: if isinstance(segment, str): rendered.append(segment) continue - result = Expression._evaluate_cel(segment.source, context) + result = Expression._evaluate_cel(segment.source, context, run_context) rendered.append("" if result is None else _stringify_cel_value(result)) return "".join(rendered) @staticmethod - def _evaluate_cel(expression: str, context: dict[str, Any]) -> Any: + def _evaluate_cel( + expression: str, context: dict[str, Any], run_context: _CelRunContext + ) -> Any: try: - from celpy import Environment from celpy.adapter import CELJSONEncoder, json_to_cel from celpy.evaluation import Context - environment = Environment() + environment = _cel_environment() program = environment.program( - Expression._compile_cel(expression, environment=environment) + Expression._compile_cel(expression, environment=environment), + functions=_cel_functions(run_context), ) result = program.evaluate(cast(Context, json_to_cel(context))) if (eval_error := _find_cel_eval_error(result)) is not None: @@ -371,9 +440,7 @@ class Expression: environment: Any | None = None, ) -> Any: if environment is None: - from celpy import Environment - - environment = Environment() + environment = _cel_environment() try: return environment.compile(expression) except Exception as e: diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py index 8c756a39d..63d9630d6 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/lib/crewai/src/crewai/flow/runtime/__init__.py @@ -13,7 +13,7 @@ from collections.abc import Callable, Iterator, Sequence from concurrent.futures import Future, ThreadPoolExecutor import contextvars import copy -from datetime import datetime +from datetime import datetime, timezone import enum import inspect import logging @@ -772,6 +772,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): # duration span emitted at the end does not need to hold a span open for # the life of the run. _telemetry_started_at: float | None = PrivateAttr(default=None) + _cel_now: datetime | None = PrivateAttr(default=None) _event_futures: list[Future[None]] = PrivateAttr(default_factory=list) _pending_feedback_context: PendingFeedbackContext | None = PrivateAttr(default=None) _human_feedback_method_outputs: dict[str, Any] = PrivateAttr(default_factory=dict) @@ -1382,6 +1383,10 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): "No pending feedback context. Use from_pending() to restore a paused flow." ) + # A fresh instant, not the persisted kickoff one: a flow can pause on + # feedback for days, and expressions after resume must see today. + self._cel_now = datetime.now(timezone.utc) + execution_token = begin_execution(self._pending_feedback_context.execution_uuid) # Force `current_flow_id` to this flow's match id for the @@ -2171,6 +2176,8 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): restore_from_state_id=restore_from_state_id, ) + self._cel_now = datetime.now(timezone.utc) + ctx = baggage.set_baggage("flow_inputs", inputs or {}) ctx = baggage.set_baggage("flow_input_files", input_files or {}, context=ctx) flow_token = attach(ctx) diff --git a/lib/crewai/tests/test_flow_from_definition.py b/lib/crewai/tests/test_flow_from_definition.py index 14b85fedc..1cc05cac0 100644 --- a/lib/crewai/tests/test_flow_from_definition.py +++ b/lib/crewai/tests/test_flow_from_definition.py @@ -3047,6 +3047,94 @@ def test_expression_keeps_short_circuited_cel_errors(): ) +def test_expression_now_evaluates_with_frozen_timestamp(): + from datetime import datetime, timezone + + from crewai.flow.expressions import Expression + + frozen = datetime(2026, 1, 15, 12, 30, tzinfo=timezone.utc) + + assert Expression("now().getFullYear()", context={}, now=frozen).evaluate() == 2026 + assert ( + Expression("string(now())", context={}, now=frozen).evaluate() + == "2026-01-15T12:30:00Z" + ) + assert ( + Expression("string(now() - duration('24h'))", context={}, now=frozen).evaluate() + == "2026-01-14T12:30:00Z" + ) + + +def test_expression_now_defaults_to_current_time(): + from datetime import datetime, timezone + + from crewai.flow.expressions import Expression + + year = Expression("now().getFullYear()", context={}).evaluate() + + assert year == datetime.now(timezone.utc).year + + +def test_expression_now_renders_in_templates(): + from datetime import datetime, timezone + + from crewai.flow.expressions import Expression + + frozen = datetime(2026, 1, 15, tzinfo=timezone.utc) + rendered = Expression( + {"query": "News from ${string(now().getFullYear())}"}, + context={}, + now=frozen, + ).render_template() + + assert rendered == {"query": "News from 2026"} + + +def test_expression_now_passes_root_validation(): + from crewai.flow.expressions import Expression + + Expression("string(now().getFullYear())").validate_expression( + allowed_roots=["state", "outputs"] + ) + + +def test_expression_from_flow_uses_run_frozen_now(): + from datetime import datetime, timezone + + from crewai.flow.expressions import Expression + + flow = Flow() + flow._cel_now = datetime(2026, 1, 15, tzinfo=timezone.utc) + + assert ( + Expression.from_flow("now().getFullYear()", flow).evaluate() == 2026 + ) + + +def test_expression_action_can_use_now(): + definition = FlowDefinition.from_declaration(contents= + { + "schema": "crewai.flow/v1", + "name": "NowFlow", + "methods": { + "today": { + "start": True, + "do": { + "call": "expression", + "expr": "string(now().getFullYear())", + }, + } + }, + } + ) + + from datetime import datetime, timezone + + result = Flow.from_declaration(contents=definition).kickoff() + + assert result == str(datetime.now(timezone.utc).year) + + def test_expression_action_can_route_like_if_else(): yaml_str = f""" schema: crewai.flow/v1 @@ -3738,6 +3826,41 @@ def test_resume_synthetic_completion_persists(): assert _saved_methods("resume-synthetic") == ["generate"] +def test_resume_freezes_fresh_cel_now(): + from crewai.flow.expressions import Expression + + backend = DefinitionStoreBackend(store="resume-cel-now") + frozen_at_listener: list[Any] = [] + + class NowResumableFlow(Flow): + @start() + @human_feedback(message="Review:") + def generate(self): + return "content" + + @listen(generate) + def process(self, result): + frozen_at_listener.append(self._cel_now) + return Expression.from_flow("string(now())", self).evaluate() + + context = PendingFeedbackContext( + flow_id="resume-cel-now-1", + flow_class="NowResumableFlow", + method_name="generate", + method_output="content", + message="Review:", + ) + backend.save_pending_feedback("resume-cel-now-1", context, {"id": "resume-cel-now-1"}) + + flow = NowResumableFlow.from_pending("resume-cel-now-1", backend) + assert flow._cel_now is None + + result = flow.resume("looks good") + + assert frozen_at_listener[0] is not None + assert result == frozen_at_listener[0].strftime("%Y-%m-%dT%H:%M:%SZ") + + class ReviewFlow(Flow): @start() @human_feedback(