feat: add PluginResult.wait_for_background_tasks()

Signed-off-by: Alex Bozarth <ajbozart@us.ibm.com>

Author: ajbozarth

cpex/framework/manager.py

@@ -769,9 +769,10 @@ async def _run_fire_and_forget_task(
         local_context: PluginContext,
         semaphore: Optional[asyncio.Semaphore],
         extensions: Optional[Extensions] = None,
-    ) -> None:
+    ) -> Optional[PluginErrorModel]:
         """Execute a plugin as a fire-and-forget background task.
 
+        Returns None on success, or a PluginErrorModel if the plugin raised.
         Errors are logged but never propagated — background tasks cannot halt the pipeline.
         If on_error=DISABLE, the plugin is added to the runtime-disabled set.
         """
@@ -781,11 +782,13 @@ async def _run_fire_and_forget_task(
                     await self._execute_with_timeout(hook_ref, payload, local_context, extensions=extensions)
             else:
                 await self._execute_with_timeout(hook_ref, payload, local_context, extensions=extensions)
-        except Exception:
+            return None
+        except Exception as exc:
             logger.error("Plugin %s failed in fire-and-forget mode (ignored)", hook_ref.plugin_ref.name)
             if hook_ref.plugin_ref.on_error == OnError.DISABLE:
                 self._runtime_disabled.add(hook_ref.plugin_ref.name)
             # FAIL and IGNORE both just log for FIRE_AND_FORGET mode (background can't halt pipeline)
+            return PluginErrorModel(message=repr(exc), plugin_name=hook_ref.plugin_ref.name)
 
     async def execute_plugin(
         self,

cpex/framework/models.py

@@ -1500,10 +1500,8 @@ class PluginResult(BaseModel, Generic[T]):
             violation (Optional[PluginViolation]): violation object.
             metadata (Optional[dict[str, Any]]): additional metadata.
             background_tasks (list[asyncio.Task]): asyncio.Task handles for any FIRE_AND_FORGET
-                plugins scheduled during this invocation. Use
-                ``await asyncio.gather(*result.background_tasks, return_exceptions=True)``
-                to deterministically wait for all background tasks to complete (useful in tests).
-                This field is excluded from model serialization.
+                plugins scheduled during this invocation. Use ``wait_for_background_tasks()``
+                to await them and collect any errors. This field is excluded from model serialization.
 
      Examples:
         >>> result = PluginResult()
@@ -1537,6 +1535,20 @@ class PluginResult(BaseModel, Generic[T]):
     metadata: Optional[dict[str, Any]] = Field(default_factory=dict)
     background_tasks: list[asyncio.Task] = Field(default_factory=list, exclude=True)
 
+    async def wait_for_background_tasks(self) -> "list[PluginErrorModel]":
+        """Await all FIRE_AND_FORGET background tasks and return any errors.
+
+        Returns an empty list if all tasks completed without error.
+
+        Examples:
+            >>> result = PluginResult()
+            >>> # errors = await result.wait_for_background_tasks()
+        """
+        if not self.background_tasks:
+            return []
+        results = await asyncio.gather(*self.background_tasks, return_exceptions=True)
+        return [r for r in results if isinstance(r, PluginErrorModel)]
+
 
 class GlobalContext(BaseModel):
     """The global context, which shared across all plugins.

docs/specs/plugin-framework-spec.md

@@ -188,11 +188,11 @@ class PluginResult(Generic[T]):
     background_tasks: list[asyncio.Task] = []  # excluded from serialization
 ```
 
-`background_tasks` contains the `asyncio.Task` handles for any `FIRE_AND_FORGET` plugins scheduled during the invocation. Use it to wait for background tasks without sleep delays:
+`background_tasks` contains the `asyncio.Task` handles for any `FIRE_AND_FORGET` plugins scheduled during the invocation. Use `wait_for_background_tasks()` to await them and collect any errors:
 
 ```python
 result, _ = await manager.invoke_hook(...)
-await asyncio.gather(*result.background_tasks, return_exceptions=True)
+errors = await result.wait_for_background_tasks()  # list[PluginErrorModel], empty on success
 ```
 
 The `PluginViolation` type carries structured policy failure information:

tests/unit/cpex/framework/test_plugin_modes.py

@@ -102,7 +102,7 @@ async def prompt_pre_fetch(self, payload, context):
     assert not finished.is_set()
 
     # Wait deterministically for the background task to complete
-    await asyncio.gather(*result.background_tasks, return_exceptions=True)
+    await result.wait_for_background_tasks()
     assert finished.is_set()
 
     await manager.shutdown()
@@ -129,8 +129,35 @@ async def prompt_pre_fetch(self, payload, context):
 
     assert result.continue_processing
 
-    # Wait deterministically for the background task to run and silently fail
-    await asyncio.gather(*result.background_tasks, return_exceptions=True)
+    # Wait for the background task; errors are returned, not raised
+    await result.wait_for_background_tasks()
+
+    await manager.shutdown()
+
+
+@pytest.mark.asyncio
+async def test_wait_for_background_tasks_returns_errors():
+    """wait_for_background_tasks() returns a PluginErrorModel for each failed task."""
+
+    class BrokenPlugin(Plugin):
+        async def prompt_pre_fetch(self, payload, context):
+            raise RuntimeError("boom")
+
+    manager = await _make_manager()
+    cfg = make_plugin_config("BrokenFnF", PluginMode.FIRE_AND_FORGET)
+    plugin = BrokenPlugin(cfg)
+
+    with patch.object(manager._registry, "get_hook_refs_for_hook") as mock_get:
+        mock_get.return_value = [HookRef(PromptHookType.PROMPT_PRE_FETCH, PluginRef(plugin))]
+        payload = PromptPrehookPayload(prompt_id="test", args={})
+        global_context = GlobalContext(request_id="wait_errors")
+
+        result, _ = await manager.invoke_hook(PromptHookType.PROMPT_PRE_FETCH, payload, global_context)
+
+    errors = await result.wait_for_background_tasks()
+    assert len(errors) == 1
+    assert errors[0].plugin_name == "BrokenFnF"
+    assert "RuntimeError" in errors[0].message
 
     await manager.shutdown()
 
@@ -357,7 +384,7 @@ async def prompt_pre_fetch(self, payload, context):
     assert result.continue_processing
 
     # Wait deterministically for all background FIRE_AND_FORGET tasks to complete
-    await asyncio.gather(*result.background_tasks, return_exceptions=True)
+    await result.wait_for_background_tasks()
 
     # With pool=1, max concurrency should be 1
     assert concurrency_high_water <= 1
@@ -620,7 +647,7 @@ async def prompt_pre_fetch(self, payload, context):
 
     assert result.continue_processing
     # F&F is async — wait for it deterministically
-    await asyncio.gather(*result.background_tasks, return_exceptions=True)
+    await result.wait_for_background_tasks()
 
     assert phase_log == ["seq", "xform", "audit", "conc", "fnf"]
 
@@ -666,7 +693,7 @@ async def prompt_pre_fetch(self, payload, context):
     # FIRE_AND_FORGET has not yet completed (fire-and-forget)
     assert not fire_and_forget_started.is_set()
 
-    await asyncio.gather(*result.background_tasks, return_exceptions=True)
+    await result.wait_for_background_tasks()
     assert "fire_and_forget" in phase_log
     # FIRE_AND_FORGET always comes after sequential in the log
     assert phase_log.index("sequential") < phase_log.index("fire_and_forget")