fix: prevent orphan SUBMITTED rows when producer fails early

src/a2a/server/agent_execution/active_task.py

@@ -558,6 +558,16 @@ async def _run_producer(self) -> None:
                     request_context.context_id or '',
                 )
                 self._task_created.set()
+                # Explicitly transition to FAILED state to avoid orphan SUBMITTED rows
+                # when the consumer may not process the exception event.
+                failed_status = TaskStatus(
+                    state=TaskState.TASK_STATE_FAILED,
+                )
+                failed_event = TaskStatusUpdateEvent(
+                    task_id=self._task_id,
+                    status=failed_status,
+                )
+                await self._task_manager.save_task_event(failed_event)
             await self._event_queue_agent.enqueue_event(cast('Event', e))
 
         finally:

src/a2a/server/agent_execution/active_task_registry.py

@@ -11,6 +11,7 @@
     from a2a.server.context import ServerCallContext
     from a2a.server.tasks.push_notification_sender import PushNotificationSender
     from a2a.server.tasks.task_store import TaskStore
+    from a2a.types.a2a_pb2 import Message
 
 from a2a.server.agent_execution.active_task import ActiveTask
 from a2a.server.tasks.task_manager import TaskManager
@@ -41,6 +42,7 @@ async def get_or_create(
         call_context: ServerCallContext,
         context_id: str | None = None,
         create_task_if_missing: bool = False,
+        initial_message: Message | None = None,
     ) -> ActiveTask:
         """Retrieves an existing ActiveTask or creates a new one."""
         async with self._lock:
@@ -51,7 +53,7 @@ async def get_or_create(
                 task_id=task_id,
                 context_id=context_id,
                 task_store=self._task_store,
-                initial_message=None,
+                initial_message=initial_message,
                 context=call_context,
             )
 

src/a2a/server/request_handlers/default_request_handler_v2.py

@@ -222,6 +222,7 @@ async def _setup_active_task(
             context_id=context_id,
             call_context=call_context,
             create_task_if_missing=True,
+            initial_message=params.message,
         )
 
         return active_task, request_context

tests/server/agent_execution/test_active_task.py

@@ -895,3 +895,57 @@ async def execute_mock(req, q):
     assert len(events) == 0
 
     await active_task.cancel(request_context)
+
+
+@pytest.mark.asyncio
+async def test_active_task_producer_exception_sets_failed_state():
+    """Test that producer exception explicitly sets FAILED state.
+
+    Regression test for #1067: ActiveTaskRegistry produces orphan SUBMITTED rows
+    when AgentExecutor.execute() raises before emitting any events.
+    """
+    agent_executor = Mock()
+    task_manager = Mock()
+    request_context = Mock(spec=RequestContext)
+    request_context.context_id = 'test-context-id'
+    request_context.call_context = ServerCallContext()
+
+    active_task = ActiveTask(
+        agent_executor=agent_executor,
+        task_id='test-task-id',
+        task_manager=task_manager,
+        push_sender=Mock(),
+    )
+
+    # Simulate executor raising an exception before emitting any events
+    async def execute_mock(req, q):
+        raise RuntimeError('Executor failed before emitting events')
+
+    agent_executor.execute = AsyncMock(side_effect=execute_mock)
+    agent_executor.cancel = AsyncMock()
+    task_manager.get_task = AsyncMock(return_value=None)
+    task_manager.ensure_task_id = AsyncMock(
+        return_value=Task(
+            id='test-task-id',
+            status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED),
+        )
+    )
+    task_manager.save_task_event = AsyncMock()
+    task_manager.process = AsyncMock(side_effect=lambda x: x)
+
+    await active_task.enqueue_request(request_context)
+    await active_task.start(
+        call_context=ServerCallContext(), create_task_if_missing=True
+    )
+
+    # Wait for producer to finish
+    await asyncio.sleep(0.2)
+
+    # Verify that save_task_event was called with a FAILED status
+    assert task_manager.save_task_event.called
+    # Get the last call to save_task_event
+    last_call = task_manager.save_task_event.call_args
+    if last_call:
+        event = last_call.args[0]
+        assert isinstance(event, TaskStatusUpdateEvent)
+        assert event.status.state == TaskState.TASK_STATE_FAILED