Python: Workflow event source updates (#789)

* Workflow event source updates

* Add WorkflowLifecycleEvent TypeAlias. Update docstrings

* Updates

* Rename

---------

Co-authored-by: Chris <66376200+crickman@users.noreply.github.com>

python/packages/main/tests/workflow/test_runner.py

@@ -5,10 +5,19 @@ from dataclasses import dataclass
 
 import pytest
 
-from agent_framework import Executor, WorkflowCompletedEvent, WorkflowContext, WorkflowEvent, handler
+from agent_framework import (
+    AgentExecutorResponse,
+    AgentRunResponse,
+    Executor,
+    WorkflowCompletedEvent,
+    WorkflowContext,
+    WorkflowEvent,
+    WorkflowEventSource,
+    handler,
+)
 from agent_framework._workflow._edge import SingleEdgeGroup
 from agent_framework._workflow._runner import Runner
-from agent_framework._workflow._runner_context import InProcRunnerContext, RunnerContext
+from agent_framework._workflow._runner_context import InProcRunnerContext, Message, RunnerContext
 from agent_framework._workflow._shared_state import SharedState
 
 
@@ -79,6 +88,7 @@ async def test_runner_run_until_convergence():
         assert isinstance(event, WorkflowEvent)
         if isinstance(event, WorkflowCompletedEvent):
             result = event.data
+            assert event.origin is WorkflowEventSource.EXECUTOR
 
     assert result is not None and result == 10
 
@@ -148,3 +158,20 @@ async def test_runner_already_running():
                 pass
 
         await asyncio.gather(_run(), _run())
+
+
+async def test_runner_emits_runner_completion_for_agent_response_without_targets():
+    ctx = InProcRunnerContext()
+    runner = Runner([], {}, SharedState(), ctx)
+
+    await ctx.send_message(
+        Message(
+            data=AgentExecutorResponse("agent", AgentRunResponse()),
+            source_id="agent",
+        )
+    )
+
+    events: list[WorkflowEvent] = [event async for event in runner.run_until_convergence()]
+    completions = [e for e in events if isinstance(e, WorkflowCompletedEvent)]
+    assert completions
+    assert all(e.origin is WorkflowEventSource.FRAMEWORK for e in completions)

python/packages/main/tests/workflow/test_simple_sub_workflow.py

@@ -3,8 +3,6 @@
 import asyncio
 from dataclasses import dataclass
 
-import pytest
-
 from agent_framework import (
     Executor,
     WorkflowBuilder,
@@ -64,7 +62,6 @@ class SimpleParent(Executor):
         self.result = response
 
 
-@pytest.mark.asyncio
 async def test_simple_sub_workflow():
     """Test the simplest possible sub-workflow."""
     # Create sub-workflow with dummy executor to satisfy validation

python/packages/main/tests/workflow/test_sub_workflow.py

@@ -4,7 +4,6 @@ import asyncio
 from dataclasses import dataclass
 from typing import Any
 
-import pytest
 from pydantic import Field
 
 from agent_framework import (
@@ -120,7 +119,6 @@ class ParentOrchestrator(Executor):
         self.results.append(result)
 
 
-@pytest.mark.asyncio
 async def test_basic_sub_workflow() -> None:
     """Test basic sub-workflow execution without interception."""
     # Create sub-workflow
@@ -185,7 +183,6 @@ async def test_basic_sub_workflow() -> None:
     assert parent.result.is_valid is True
 
 
-@pytest.mark.asyncio
 async def test_sub_workflow_with_interception():
     """Test sub-workflow with parent interception of requests."""
     # Create sub-workflow
@@ -248,7 +245,6 @@ async def test_sub_workflow_with_interception():
     assert parent.results[0].reason == "Domain not approved"
 
 
-@pytest.mark.asyncio
 async def test_conditional_forwarding() -> None:
     """Test conditional forwarding with RequestResponse.forward()."""
 
@@ -326,7 +322,6 @@ async def test_conditional_forwarding() -> None:
     assert parent.result.is_valid is True
 
 
-@pytest.mark.asyncio
 async def test_workflow_scoped_interception() -> None:
     """Test interception scoped to specific sub-workflows."""
 

python/packages/main/tests/workflow/test_tracing.py

@@ -273,7 +273,6 @@ async def test_trace_context_handling(tracing_enabled: Any, span_exporter: InMem
     assert processing_span.attributes.get("message.type") == "str"
 
 
-@pytest.mark.asyncio
 async def test_trace_context_disabled_when_tracing_disabled() -> None:
     """Test that no trace context is added when tracing is disabled."""
     # Tracing should be disabled by default
@@ -299,7 +298,6 @@ async def test_trace_context_disabled_when_tracing_disabled() -> None:
     assert message.source_span_id is None
 
 
-@pytest.mark.asyncio
 async def test_end_to_end_workflow_tracing(tracing_enabled: Any, span_exporter: InMemorySpanExporter) -> None:
     """Test end-to-end tracing including workflow build, execution, and span linking with fan-in edges."""
     # Create executors for fan-in scenario
@@ -424,7 +422,6 @@ async def test_end_to_end_workflow_tracing(tracing_enabled: Any, span_exporter:
     assert len(aggregator_span.links) >= 2, f"Expected at least 2 links, got {len(aggregator_span.links)}"
 
 
-@pytest.mark.asyncio
 async def test_workflow_error_handling_in_tracing(tracing_enabled: Any, span_exporter: InMemorySpanExporter) -> None:
     """Test that workflow errors are properly recorded in traces."""
 
@@ -460,7 +457,6 @@ async def test_workflow_error_handling_in_tracing(tracing_enabled: Any, span_exp
     assert workflow_span.status.status_code.name == "ERROR"
 
 
-@pytest.mark.asyncio
 async def test_message_trace_context_serialization() -> None:
     """Test that message trace context is properly serialized/deserialized."""
     ctx = InProcRunnerContext()
@@ -495,7 +491,6 @@ async def test_message_trace_context_serialization() -> None:
     assert restored_msg.source_span_ids == ["span123"]  # Test new format
 
 
-@pytest.mark.asyncio
 async def test_workflow_build_error_tracing(tracing_enabled: Any, span_exporter: InMemorySpanExporter) -> None:
     """Test that build errors are properly recorded in build spans."""
 

python/packages/main/tests/workflow/test_workflow.py

@@ -231,7 +231,7 @@ async def test_fan_out():
 
     events = await workflow.run(NumberMessage(data=0))
 
-    # Each executor will emit two events: ExecutorInvokeEvent and ExecutorCompletedEvent
+    # Each executor will emit two events: ExecutorInvokedEvent and ExecutorCompletedEvent
     # executor_b will also emit a WorkflowCompletedEvent
     assert len(events) == 7
 
@@ -251,7 +251,7 @@ async def test_fan_out_multiple_completed_events():
 
     events = await workflow.run(NumberMessage(data=0))
 
-    # Each executor will emit two events: ExecutorInvokeEvent and ExecutorCompletedEvent
+    # Each executor will emit two events: ExecutorInvokedEvent and ExecutorCompletedEvent
     # executor_a and executor_b will also emit a WorkflowCompletedEvent
     assert len(events) == 8
 
@@ -276,7 +276,7 @@ async def test_fan_in():
 
     events = await workflow.run(NumberMessage(data=0))
 
-    # Each executor will emit two events: ExecutorInvokeEvent and ExecutorCompletedEvent
+    # Each executor will emit two events: ExecutorInvokedEvent and ExecutorCompletedEvent
     # aggregator will also emit a WorkflowCompletedEvent
     assert len(events) == 9
 
@@ -667,10 +667,10 @@ async def test_workflow_with_simple_cycle_and_exit_condition():
     )  # Should complete when executor_a reaches its limit
 
     # Verify cycling occurred (should have events from both executors)
-    # Check for ExecutorInvokeEvent and ExecutorCompletedEvent types that have executor_id
-    from agent_framework import ExecutorCompletedEvent, ExecutorInvokeEvent
+    # Check for ExecutorInvokedEvent and ExecutorCompletedEvent types that have executor_id
+    from agent_framework import ExecutorCompletedEvent, ExecutorInvokedEvent
 
-    executor_events = [e for e in events if isinstance(e, (ExecutorInvokeEvent, ExecutorCompletedEvent))]
+    executor_events = [e for e in events if isinstance(e, (ExecutorInvokedEvent, ExecutorCompletedEvent))]
     executor_ids = {e.executor_id for e in executor_events}
     assert "exec_a" in executor_ids, "Should have events from executor A"
     assert "exec_b" in executor_ids, "Should have events from executor B"

python/packages/main/tests/workflow/test_workflow_agent.py

@@ -76,7 +76,6 @@ class RequestingExecutor(Executor):
 class TestWorkflowAgent:
     """Test cases for WorkflowAgent end-to-end functionality."""
 
-    @pytest.mark.asyncio
     async def test_end_to_end_basic_workflow(self):
         """Test basic end-to-end workflow execution with 2 executors emitting AgentRunEvent."""
         # Create workflow with two executors
@@ -117,7 +116,6 @@ class TestWorkflowAgent:
         assert "Step1: Hello World" in step1_text
         assert "Step2: Step1: Hello World" in step2_text
 
-    @pytest.mark.asyncio
     async def test_end_to_end_basic_workflow_streaming(self):
         """Test end-to-end workflow with streaming executor that emits AgentRunStreamingEvent."""
         # Create a single streaming executor
@@ -146,7 +144,6 @@ class TestWorkflowAgent:
         assert isinstance(second_content, TextContent)
         assert "Streaming2: Streaming1: Test input" in second_content.text
 
-    @pytest.mark.asyncio
     async def test_end_to_end_request_info_handling(self):
         """Test end-to-end workflow with RequestInfoEvent handling."""
         # Create workflow with requesting executor -> request info executor (no cycle)

python/packages/main/tests/workflow/test_workflow_context.py

@@ -0,0 +1,64 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from typing import TYPE_CHECKING
+
+from agent_framework import (
+    WorkflowCompletedEvent,
+    WorkflowContext,
+    WorkflowEvent,
+    WorkflowRunState,
+    WorkflowStatusEvent,
+)
+
+if TYPE_CHECKING:
+    from _pytest.logging import LogCaptureFixture
+
+    from agent_framework._workflow._runner_context import InProcRunnerContext
+
+
+@asynccontextmanager
+async def make_context(
+    executor_id: str = "exec",
+) -> AsyncIterator[tuple[WorkflowContext[object], "InProcRunnerContext"]]:
+    from agent_framework._workflow._runner_context import InProcRunnerContext
+    from agent_framework._workflow._shared_state import SharedState
+
+    runner_ctx = InProcRunnerContext()
+    shared_state = SharedState()
+    workflow_ctx: WorkflowContext[object] = WorkflowContext(
+        executor_id,
+        ["source"],
+        shared_state,
+        runner_ctx,
+    )
+    try:
+        yield workflow_ctx, runner_ctx
+    finally:
+        await asyncio.sleep(0)
+
+
+async def test_executor_cannot_emit_framework_lifecycle_event(caplog: "LogCaptureFixture") -> None:
+    async with make_context() as (ctx, runner_ctx):
+        caplog.clear()
+        with caplog.at_level("WARNING"):
+            await ctx.add_event(WorkflowStatusEvent(state=WorkflowRunState.IN_PROGRESS))
+
+        events: list[WorkflowEvent] = await runner_ctx.drain_events()
+        assert len(events) == 1
+        assert type(events[0]).__name__ == "WorkflowWarningEvent"
+        data = getattr(events[0], "data", None)
+        assert isinstance(data, str)
+        assert "reserved for framework lifecycle notifications" in data
+        assert any("attempted to emit WorkflowStatusEvent" in message for message in list(caplog.messages))
+
+
+async def test_executor_emits_normal_event() -> None:
+    async with make_context() as (ctx, runner_ctx):
+        await ctx.add_event(WorkflowCompletedEvent("done"))
+
+        events: list[WorkflowEvent] = await runner_ctx.drain_events()
+        assert len(events) == 1
+        assert isinstance(events[0], WorkflowCompletedEvent)

python/packages/main/tests/workflow/test_workflow_states.py

@@ -14,9 +14,11 @@ from agent_framework import (
     WorkflowBuilder,
     WorkflowCompletedEvent,
     WorkflowContext,
+    WorkflowEventSource,
     WorkflowFailedEvent,
     WorkflowRunResult,
     WorkflowRunState,
+    WorkflowStartedEvent,
     WorkflowStatusEvent,
     handler,
 )
@@ -41,9 +43,12 @@ async def test_executor_failed_and_workflow_failed_events_streaming():
             events.append(ev)
 
     # Workflow-level failure and FAILED status should be surfaced
-    assert any(isinstance(e, WorkflowFailedEvent) for e in events)
+    failed_events = [e for e in events if isinstance(e, WorkflowFailedEvent)]
+    assert failed_events
+    assert all(e.origin is WorkflowEventSource.FRAMEWORK for e in failed_events)
     status = [e for e in events if isinstance(e, WorkflowStatusEvent)]
     assert status and status[-1].state == WorkflowRunState.FAILED
+    assert all(e.origin is WorkflowEventSource.FRAMEWORK for e in status)
 
 
 async def test_executor_failed_event_emitted_on_direct_execute():
@@ -58,7 +63,9 @@ async def test_executor_failed_event_emitted_on_direct_execute():
     with pytest.raises(RuntimeError, match="boom"):
         await failing.execute(0, wf_ctx)
     drained = await ctx.drain_events()
-    assert any(isinstance(e, ExecutorFailedEvent) for e in drained)
+    failed = [e for e in drained if isinstance(e, ExecutorFailedEvent)]
+    assert failed
+    assert all(e.origin is WorkflowEventSource.FRAMEWORK for e in failed)
 
 
 class Requester(Executor):
@@ -99,6 +106,19 @@ async def test_completed_status_streaming():
     # Last status should be COMPLETED
     status = [e for e in events if isinstance(e, WorkflowStatusEvent)]
     assert status and status[-1].state == WorkflowRunState.COMPLETED
+    assert all(e.origin is WorkflowEventSource.FRAMEWORK for e in status)
+
+
+async def test_started_and_completed_event_origins():
+    c = Completer(id="c-origin")
+    wf = WorkflowBuilder().set_start_executor(c).build()
+    events = [ev async for ev in wf.run_stream("payload")]
+
+    started = next(e for e in events if isinstance(e, WorkflowStartedEvent))
+    assert started.origin is WorkflowEventSource.FRAMEWORK
+
+    completed = next(e for e in events if isinstance(e, WorkflowCompletedEvent))
+    assert completed.origin is WorkflowEventSource.EXECUTOR
 
 
 async def test_non_streaming_final_state_helpers():