Merge branch 'main' into heartbeat

Author: timl3136

.github/workflows/ci_checks.yml

@@ -51,10 +51,10 @@ jobs:
         uses: astral-sh/setup-uv@v1
       - name: Install dependencies
         run: |
-          uv sync --extra dev
+          uv sync --extra dev --extra openai
       - name: Run mypy type checker
         run: |
-          uv tool run mypy cadence/
+          uv run mypy cadence/
 
   test:
     name: Unit Tests

cadence/_internal/activity/_definition.py

@@ -14,7 +14,9 @@
     ParamSpec,
     TypeVar,
     Awaitable,
+    Type,
     cast,
+    overload,
     Concatenate,
 )
 
@@ -119,10 +121,13 @@ def __init__(
         super().__init__(name, wrapped, ExecutionStrategy.THREAD_POOL, signature)
         update_wrapper(self, wrapped)
 
-    def __get__(self, instance, owner):
+    @overload
+    def __get__(self, instance: None, owner: Type[T]) -> "SyncMethodImpl[T, P, R]": ...
+    @overload
+    def __get__(self, instance: T, owner: Type[T]) -> SyncImpl[P, R]: ...
+    def __get__(self, instance: T | None, owner: Type[T]) -> "SyncImpl[P, R] | Self":
         if instance is None:
             return self
-        # If we bound the method to an instance, then drop the self parameter. It's a normal function again
         return SyncImpl[P, R](
             partial(self._wrapped, instance), self.name, self._signature
         )
@@ -181,10 +186,13 @@ def __init__(
         else:
             self._is_coroutine = _COROUTINE_MARKER
 
-    def __get__(self, instance, owner):
+    @overload
+    def __get__(self, instance: None, owner: Type[T]) -> "AsyncMethodImpl[T, P, R]": ...
+    @overload
+    def __get__(self, instance: T, owner: Type[T]) -> AsyncImpl[P, R]: ...
+    def __get__(self, instance: T | None, owner: Type[T]) -> "AsyncImpl[P, R] | Self":
         if instance is None:
             return self
-        # If we bound the method to an instance, then drop the self parameter. It's a normal function again
         return AsyncImpl[P, R](
             partial(self._wrapped, instance), self.name, self._signature
         )

cadence/_internal/workflow/workflow_engine.py

@@ -13,16 +13,19 @@
 from cadence._internal.workflow.statemachine.decision_manager import DecisionManager
 from cadence._internal.workflow.workflow_instance import WorkflowInstance
 from cadence.api.v1 import history
-from cadence.api.v1.common_pb2 import Failure
+from cadence.api.v1.common_pb2 import Failure, WorkflowType
 from cadence.api.v1.decision_pb2 import (
     Decision,
     FailWorkflowExecutionDecisionAttributes,
     CompleteWorkflowExecutionDecisionAttributes,
+    ContinueAsNewWorkflowExecutionDecisionAttributes,
 )
 from cadence.api.v1.history_pb2 import (
     HistoryEvent,
     WorkflowExecutionStartedEventAttributes,
 )
+from cadence.api.v1.tasklist_pb2 import TaskList
+from cadence.error import ContinueAsNewError
 from cadence.workflow import WorkflowDefinition, WorkflowInfo
 
 logger = logging.getLogger(__name__)
@@ -187,6 +190,25 @@ def _maybe_complete_workflow(self) -> Optional[Decision]:
                 return None
         except (CancelledError, InvalidStateError, FatalDecisionError):
             raise
+        except ContinueAsNewError as e:
+            # Use execution's workflow type and task list when not overridden
+            info = self._context.info()
+            attrs = ContinueAsNewWorkflowExecutionDecisionAttributes(
+                workflow_type=WorkflowType(name=e.workflow_type or info.workflow_type),
+                task_list=TaskList(name=e.task_list or info.workflow_task_list),
+                input=info.data_converter.to_data(list(e.workflow_args)),
+            )
+            if e.execution_start_to_close_timeout is not None:
+                attrs.execution_start_to_close_timeout.FromTimedelta(
+                    e.execution_start_to_close_timeout
+                )
+            if e.task_start_to_close_timeout is not None:
+                attrs.task_start_to_close_timeout.FromTimedelta(
+                    e.task_start_to_close_timeout
+                )
+            return Decision(
+                continue_as_new_workflow_execution_decision_attributes=attrs,
+            )
         except ExceptionGroup as e:
             if e.subgroup((InvalidStateError, FatalDecisionError)):
                 raise

cadence/contrib/__init__.py

@@ -0,0 +1,124 @@
+# OpenAI Agents SDK Integration with Cadence
+
+We welcome contributions. Join [CNCF Slack Workspace](https://communityinviter.com/apps/cloud-native/cncf) and contact us in [#cadence-users](https://cloud-native.slack.com/archives/C09J2FQ7XU3).
+
+## Durable Agent Execution
+
+The module integrates with [OpenAI Agents SDK](https://github.com/openai/openai-agents-python) to allow users to run OpenAI agents in a durable way (retries, reset to a check point).
+
+Currently, it only supports agents with function tools.
+
+**Requirement** Python version 3.12+ is required. 
+
+## Example: Booking flight Agent
+
+The example below demonstrates how to run a single agent with one tool. See full code example [here](https://gist.github.com/shijiesheng/a06b6b58e44a9a09555bd3140370700e)
+
+### Step 1: Write Agent as Cadence Workflow, tools as Cadence Activity (book_flight_agent.py)
+
+```python
+import cadence
+from agents import Agent, function_tool, Runner, RunConfig
+from cadence.contrib.openai import OpenAIActivities
+from datetime import datetime
+from dataclasses import dataclass
+
+cadence_registry = cadence.Registry()
+cadence_registry.register_activities(OpenAIActivities())
+
+@cadence_registry.workflow(name="BookFlightAgentWorkflow")
+class BookFlightAgentWorkflow:
+
+    @cadence.workflow.run
+    async def run(self, input: str) -> str:
+
+        # define agent using OpenAI SDK as usual
+        agent =Agent(
+            name = "Book Flight Agent",
+            model = "gpt-4o-mini",
+            tools = [
+                function_tool(book_flight),
+            ],
+        )
+        result = await Runner.run(agent, input, run_config=RunConfig(
+                tracing_disabled=True,
+            ))
+        return result.final_output
+
+@dataclass
+class Flight:
+    from_city: str
+    to_city: str
+    departure_date: datetime
+    return_date: datetime
+    price: float
+    airline: str
+    flight_number: str
+    seat_number: str
+
+@cadence_registry.activity(name="book_flight")
+async def book_flight(from_city: str, to_city: str, departure_date: datetime, return_date: datetime) -> Flight:
+    """
+    Book a flight tool: a pure mock for demo purposes
+    """
+    return Flight(from_city=from_city, to_city=to_city, departure_date=departure_date, return_date=return_date, price=100, airline="United", flight_number="123456", seat_number="12A")
+```
+
+### Step 2: Start Cadence Server
+
+```sh
+curl https://raw.githubusercontent.com/cadence-workflow/cadence/master/docker/docker-compose.yml | docker compose -f - up -d
+```
+
+### Step 3: Trigger the Agent Run
+```python
+# in main.py
+
+import asyncio
+import cadence
+from datetime import timedelta
+from cadence.api.v1.history_pb2 import EventFilterType
+from cadence.contrib.openai import PydanticDataConverter
+from book_flight_agent import cadence_registry
+from cadence.api.v1.service_workflow_pb2 import GetWorkflowExecutionHistoryRequest
+
+async def main():
+    # start Cadence worker
+    worker = cadence.worker.Worker(
+        cadence.Client(
+            domain="default",
+            target="localhost:7833",
+            data_converter=PydanticDataConverter(),
+        ),
+        "agent-task-list",
+        cadence_registry,
+    )
+
+    # start BookFlightAgentWorkflow
+    async with worker:
+        execution =await worker.client.start_workflow(
+            "BookFlightAgentWorkflow",
+            "Book a flight from New York to London on March 20th 2026 at 10:00 AM and return on March 25th, 2026 at 10:00 AM",
+            task_list=worker.task_list,
+            execution_start_to_close_timeout=timedelta(minutes=10),
+        )
+
+        print(f"cadence workflow started: http://localhost:8088/domains/default/cluster0/workflows/{execution.workflow_id}/{execution.run_id}/summary")
+
+        await worker.client.workflow_stub.GetWorkflowExecutionHistory(
+            GetWorkflowExecutionHistoryRequest(
+                domain=worker.client.domain,
+                workflow_execution=execution,
+                wait_for_new_event=True,
+                history_event_filter_type=EventFilterType.EVENT_FILTER_TYPE_CLOSE_EVENT,
+                skip_archival=True,
+            )
+        )
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+
+### Step 4: See Agent Run in Cadence Web
+
+![Screenshot of Agent Run in Cadence Web](images/cadence-web-agent-run.jpg)

cadence/contrib/openai/README.md

@@ -0,0 +1,15 @@
+import sys
+
+if sys.version_info < (3, 12):
+    raise ImportError(
+        "The 'openai' module requires Python 3.12 or higher due to its dependency on inspect.markcoroutinefunction"
+    )
+
+from .openai_activities import OpenAIActivities  # type: ignore[unreachable]
+from .cadence_agent_runner import CadenceAgentRunner as _CadenceAgentRunner  # noqa: F401 — imported for side effect
+from .pydantic_data_converter import PydanticDataConverter
+
+__all__ = [
+    "PydanticDataConverter",
+    "OpenAIActivities",
+]

cadence/contrib/openai/__init__.py

@@ -0,0 +1,133 @@
+import dataclasses
+import sys
+from typing import Any, Unpack
+from typing_extensions import override
+from agents import (
+    Agent,
+    Handoff,
+    RunConfig,
+    RunResult,
+    RunResultStreaming,
+    RunState,
+    TContext,
+    TResponseInputItem,
+)
+from agents.run import (
+    DEFAULT_MAX_TURNS,
+    AgentRunner,
+    RunOptions,
+    set_default_agent_runner,
+)
+from cadence.contrib.openai.cadence_model import CadenceModel
+
+# TraceCtxManager assumes that the model invocation is running in the same context.
+# This is no longer true with Cadence workflows, where each invocation is running in a separate coroutine.
+# Thus, suppress the error
+_original_unraisablehook = sys.unraisablehook
+
+
+def _suppress_ctx_error_hook(args) -> None:
+    if isinstance(args.exc_value, ValueError) and (
+        isinstance(args.exc_value, ValueError)
+        and "was created in a different Context" in str(args.exc_value)
+    ):
+        return
+    _original_unraisablehook(args)
+
+
+sys.unraisablehook = _suppress_ctx_error_hook
+
+
+def _replace_model_in_agent(
+    agent: Agent[Any],
+) -> Agent[Any]:
+    if isinstance(agent.model, CadenceModel):
+        return agent
+
+    name = agent.model
+    if not isinstance(name, str):
+        raise ValueError("Model name must be a string")
+
+    agent.model = CadenceModel(model_name=name)
+
+    for i, handoff in enumerate(agent.handoffs):
+        if isinstance(handoff, Agent):
+            agent.handoffs[i] = _replace_model_in_agent(handoff)
+        elif isinstance(handoff, Handoff):
+            raise ValueError("Handoff is not yet supported")
+        else:
+            raise TypeError(f"Unknown handoff type: {type(handoff)}")
+
+    return agent
+
+
+class CadenceAgentRunner(AgentRunner):
+    @override
+    async def run(
+        self,
+        starting_agent: Agent[TContext],
+        input: str | list[TResponseInputItem] | RunState[TContext],
+        **kwargs: Unpack[RunOptions[TContext]],
+    ) -> RunResult:
+
+        context = kwargs.get("context")
+        max_turns = kwargs.get("max_turns", DEFAULT_MAX_TURNS)
+        hooks = kwargs.get("hooks")
+        run_config = kwargs.get("run_config")
+        error_handlers = kwargs.get("error_handlers")
+        previous_response_id = kwargs.get("previous_response_id")
+        auto_previous_response_id = kwargs.get("auto_previous_response_id", False)
+        conversation_id = kwargs.get("conversation_id")
+        session = kwargs.get("session")
+
+        # if starting_agent.tools:
+        #     raise ValueError(f"Tools are not yet supported")
+
+        if starting_agent.mcp_servers:
+            raise ValueError("MCP servers are not yet supported")
+
+        if not run_config:
+            run_config = RunConfig()
+
+        if run_config.model:
+            if not isinstance(run_config.model, str):
+                raise ValueError("Model must be a string")
+            run_config = dataclasses.replace(
+                run_config,
+                model=CadenceModel(run_config.model),
+            )
+
+        return await super().run(
+            starting_agent=_replace_model_in_agent(starting_agent),
+            input=input,
+            context=context,
+            max_turns=max_turns,
+            hooks=hooks,
+            run_config=run_config,
+            error_handlers=error_handlers,
+            previous_response_id=previous_response_id,
+            auto_previous_response_id=auto_previous_response_id,
+            conversation_id=conversation_id,
+            session=session,
+        )
+
+    @override
+    def run_sync(
+        self,
+        starting_agent: Agent[TContext],
+        input: str | list[TResponseInputItem] | RunState[TContext],
+        **kwargs: Unpack[RunOptions[TContext]],
+    ) -> RunResult:
+        raise RuntimeError("Model run_sync is not yet supported.")
+
+    @override
+    def run_streamed(
+        self,
+        starting_agent: Agent[TContext],
+        input: str | list[TResponseInputItem] | RunState[TContext],
+        **kwargs: Unpack[RunOptions[TContext]],
+    ) -> RunResultStreaming:
+        raise RuntimeError("Model run_streamed is not yet supported.")
+
+
+set_default_agent_runner(CadenceAgentRunner())