[NET-1341] update: Add before each and after each support for hooks and introduced postscript_failed status

CHANGELOG.md

@@ -24,6 +24,12 @@ The format is based on Keep a Changelog and this project adheres to Semantic Ver
 
 - **Refactor stream wrapper architecture to use callback injection** - `stream_utils` is now a pure utility module with no Netra-internal imports. The commit logic (serialize and set attribute on root span) is injected as a callback from `SessionManager`, eliminating the circular dependency between `stream_utils` and `SessionManager`.
 
+## [0.1.97] -
+
+- **Add simulation lifecycle hooks** - Prescript/postscript support for multi-turn simulations via `SimulationHooks` (`before_all`, `before`, `after`, `after_all`). Hooks can return setup context passed into `BaseTask.run`, and the run uses a two-phase initialize / first-turn flow so hooks execute before any LLM spend. `before_all` failure aborts the run as `prescript_failed`; item `before` failure marks only that scenario; `after` / `after_all` failures are logged and do not affect status.
+- **Add simulation `before_each` / `after_each` hooks** - Per-item lifecycle hooks that run for every dataset item. Execution order is `before_all` → `before_each` → item-specific `before` → task → item-specific `after` → `after_each` → `after_all`.
+- **Use explicit `.description` for simulation hook metadata** - Hook descriptions sent to the backend now come from an explicit `.description` attribute on each hook function (aligned with the TypeScript SDK), instead of reading Python docstrings.
+
 ## [0.1.96] - 2026-07-23
 
 - **Reparent children of blocked root instruments instead of dropping the subtree** - When an instrumentation is not allowed to emit root-level spans, its children are now re-parented onto the nearest valid ancestor rather than dropping the entire subtree, so downstream spans are preserved.

netra/simulation/api.py

@@ -11,8 +11,10 @@
     SimulationHooks,
     run_after,
     run_after_all,
+    run_after_each,
     run_before,
     run_before_all,
+    run_before_each,
 )
 from netra.simulation.models import ConversationStatus, FileData, SimulationItem
 from netra.simulation.task import BaseTask
@@ -73,17 +75,23 @@ def run_simulation(
             max_turns: Maximum conversation turns per item before aborting
                 (default: 50).
             hooks: Optional :class:`~netra.simulation.hooks.SimulationHooks`
-                with ``before_all``, ``before``, ``after``, and ``after_all``
-                callables. Scripts live entirely on the SDK side; only
-                lightweight metadata is forwarded to the backend for UI display.
+                with ``before_all``, ``before_each``, ``before``, ``after``,
+                ``after_each``, and ``after_all`` callables. Scripts live
+                entirely on the SDK side; only lightweight metadata is
+                forwarded to the backend for UI display.
 
                 - ``before_all`` runs once before any scenario. If it raises,
                   the entire run is marked failed and no scenarios execute.
+                - ``before_each`` runs before every scenario. If it raises,
+                  that scenario is marked ``prescript_failed`` and skipped;
+                  other scenarios continue.
                 - ``before`` runs before each scenario. If it raises, that
                   scenario is marked ``prescript_failed`` and skipped; other
                   scenarios continue.
-                - ``after`` / ``after_all`` failures are logged but do not
-                  affect scenario or run status.
+                - ``after`` / ``after_each`` failures mark that scenario as
+                  ``postscript_failed``; evaluations continue normally.
+                - ``after_all`` failure marks all scenarios as
+                  ``postscript_failed``; evaluations continue normally.
 
         Returns:
             Dictionary with simulation results, or None on failure.
@@ -151,28 +159,31 @@ def run_simulation(
         setup_contexts: dict[str, Optional[dict[str, Any]]] = {}
         failed_items: list[dict[str, Any]] = []
 
-        async def run_before_and_generate_first_turn(
-            item: dict[str, str],
-            loop: asyncio.AbstractEventLoop,
-            executor: concurrent.futures.ThreadPoolExecutor,
-        ) -> Optional[SimulationItem]:
-            """Execute the before hook and generate the first turn for a single item.
+        def _setup_item_in_thread(item: dict[str, str]) -> Optional[SimulationItem]:
+            """Run hooks + first-turn generation for one item in a dedicated thread.
 
-            The before hook runs on the event loop (it may be async). The
-            ``generate_first_turn`` HTTP call is offloaded to *executor* so
-            it does not block the loop.
-
-            Returns:
-                SimulationItem if successful, None if failed (failure recorded in failed_items).
+            Each thread gets its own event loop via ``run_async_safely`` so
+            sync and async hooks both work without blocking the orchestrator.
+            Concurrency is bounded by the enclosing ThreadPoolExecutor.
             """
+            return run_async_safely(_setup_item_async(item))
+
+        async def _setup_item_async(item: dict[str, str]) -> Optional[SimulationItem]:
+            """Async body executed inside its own event loop per thread."""
             run_item_id = item["test_run_item_id"]
             dataset_item_id = item["dataset_item_id"]
 
             has_before_hook = hooks and hooks.before and dataset_item_id in hooks.before
-            if has_before_hook:
+            has_before_each_hook = hooks and hooks.before_each is not None
+
+            # Track the furthest successfully built context so teardown can
+            # clean up anything before_each created even if before fails.
+            setup_context: Optional[dict[str, Any]] = shared_context
+            if has_before_each_hook or has_before_hook:
                 try:
-                    item_context = await run_before(hooks, dataset_item_id, shared_context)
-                    setup_contexts[run_item_id] = item_context
+                    setup_context = await run_before_each(hooks, dataset_item_id, setup_context)
+                    setup_context = await run_before(hooks, dataset_item_id, setup_context)
+                    setup_contexts[run_item_id] = setup_context
                 except Exception as exc:
                     error_msg = f"before hook failed: {exc}"
                     logger.error(
@@ -194,16 +205,14 @@ async def run_before_and_generate_first_turn(
                         "error": error_msg,
                     }
                     failed_items.append(item_result)
-                    await run_after(hooks, dataset_item_id, item_result, shared_context)
+                    await self._run_after_hooks(run_id, run_item_id, dataset_item_id, hooks, item_result, setup_context)
                     return None
             else:
-                setup_contexts[run_item_id] = shared_context
+                setup_contexts[run_item_id] = setup_context
 
-            sim_item = await loop.run_in_executor(
-                executor,
-                self._client.generate_first_turn,
-                run_id,
-                run_item_id,
+            sim_item = self._client.generate_first_turn(
+                run_id=run_id,
+                run_item_id=run_item_id,
             )
             if sim_item is None:
                 logger.warning(
@@ -222,18 +231,19 @@ async def run_before_and_generate_first_turn(
                     "error": "Failed to generate first user message",
                 }
                 failed_items.append(item_result)
-                await run_after(hooks, dataset_item_id, item_result, setup_contexts[run_item_id])
+                await self._run_after_hooks(
+                    run_id, run_item_id, dataset_item_id, hooks, item_result, setup_contexts[run_item_id]
+                )
                 return None
             return sim_item
 
-        async def _run_all_before_and_first_turns() -> list[Optional[SimulationItem]]:
+        async def _run_all_setup() -> list[Optional[SimulationItem]]:
             loop = asyncio.get_running_loop()
             with concurrent.futures.ThreadPoolExecutor(max_workers=max_concurrency) as executor:
-                return list(
-                    await asyncio.gather(*[run_before_and_generate_first_turn(item, loop, executor) for item in items])
-                )
+                futures = [loop.run_in_executor(executor, _setup_item_in_thread, item) for item in items]
+                return list(await asyncio.gather(*futures))
 
-        setup_results = run_async_safely(_run_all_before_and_first_turns())
+        setup_results = run_async_safely(_run_all_setup())
         simulation_items = [item for item in setup_results if item is not None]
 
         if not simulation_items:
@@ -244,7 +254,18 @@ async def _run_all_before_and_first_turns() -> list[Optional[SimulationItem]]:
                 "failed": failed_items,
                 "total_items": len(items),
             }
-            run_async_safely(run_after_all(hooks, results, shared_context))
+            try:
+                run_async_safely(run_after_all(hooks, results, shared_context))
+            except Exception as exc:
+                error_msg = f"after_all hook failed: {exc}"
+                logger.error("%s: %s", LOG_PREFIX, error_msg, exc_info=True)
+                for item in items:
+                    self._client.report_failure(
+                        run_id=run_id,
+                        run_item_id=item["test_run_item_id"],
+                        error=error_msg,
+                        status="postscript_failed",
+                    )
             self._client.post_run_status(run_id, "completed")
             return results
 
@@ -371,7 +392,21 @@ async def process_item(run_item: SimulationItem) -> None:
             results.setdefault("failed", []).extend(setup_failed_items)
 
         # --- after_all ---
-        await run_after_all(hooks, results, shared_context)
+        try:
+            await run_after_all(hooks, results, shared_context)
+        except Exception as exc:
+            error_msg = f"after_all hook failed: {exc}"
+            logger.error("%s: %s", LOG_PREFIX, error_msg, exc_info=True)
+            all_item_ids = [item.run_item_id for item in simulation_items]
+            if setup_failed_items:
+                all_item_ids.extend(item["run_item_id"] for item in setup_failed_items if "run_item_id" in item)
+            for rid in all_item_ids:
+                self._client.report_failure(
+                    run_id=run_id,
+                    run_item_id=rid,
+                    error=error_msg,
+                    status="postscript_failed",
+                )
 
         logger.info(
             "%s: Completed=%d, Failed=%d",
@@ -381,6 +416,35 @@ async def process_item(run_item: SimulationItem) -> None:
         )
         return results
 
+    async def _run_after_hooks(
+        self,
+        run_id: str,
+        run_item_id: str,
+        dataset_item_id: str,
+        hooks: Optional[SimulationHooks],
+        item_result: dict[str, Any],
+        setup_context: Optional[dict[str, Any]],
+    ) -> None:
+        """Run after/after_each hooks and report postscript_failed on error."""
+        try:
+            await run_after(hooks, dataset_item_id, item_result, setup_context)
+            await run_after_each(hooks, dataset_item_id, item_result, setup_context)
+        except Exception as exc:
+            error_msg = f"after hook failed: {exc}"
+            logger.error(
+                "%s: %s for run_item_id=%s",
+                LOG_PREFIX,
+                error_msg,
+                run_item_id,
+                exc_info=True,
+            )
+            self._client.report_failure(
+                run_id=run_id,
+                run_item_id=run_item_id,
+                error=error_msg,
+                status="postscript_failed",
+            )
+
     async def _execute_conversation(
         self,
         run_id: str,
@@ -454,7 +518,7 @@ async def _execute_conversation(
                         "error": error_msg,
                         "turn_id": turn_id,
                     }
-                    await run_after(hooks, dataset_item_id, item_result, setup_context)
+                    await self._run_after_hooks(run_id, run_item_id, dataset_item_id, hooks, item_result, setup_context)
                     return item_result
 
                 if response.decision == ConversationStatus.STOP:
@@ -469,7 +533,7 @@ async def _execute_conversation(
                         "success": True,
                         "final_turn_id": turn_id,
                     }
-                    await run_after(hooks, dataset_item_id, item_result, setup_context)
+                    await self._run_after_hooks(run_id, run_item_id, dataset_item_id, hooks, item_result, setup_context)
                     return item_result
 
                 message = response.next_user_message  # type:ignore[assignment]
@@ -492,7 +556,7 @@ async def _execute_conversation(
                     "error": error_msg,
                     "turn_id": turn_id,
                 }
-                await run_after(hooks, dataset_item_id, item_result, setup_context)
+                await self._run_after_hooks(run_id, run_item_id, dataset_item_id, hooks, item_result, setup_context)
                 return item_result
 
         error_msg = f"Exceeded maximum turns ({max_turns}) for run_item_id={run_item_id}"
@@ -504,5 +568,5 @@ async def _execute_conversation(
             "error": error_msg,
             "turn_id": turn_id,
         }
-        await run_after(hooks, dataset_item_id, item_result, setup_context)
+        await self._run_after_hooks(run_id, run_item_id, dataset_item_id, hooks, item_result, setup_context)
         return item_result

netra/simulation/client.py

@@ -279,7 +279,9 @@ def report_failure(
             error: Error message describing the failure.
             status: The run status to set on the item. Use ``"prescript_failed"``
                 when the failure originated from a ``before_all`` or ``before``
-                hook. Defaults to ``"failed"``.
+                hook, or ``"postscript_failed"`` when the failure originated
+                from an ``after``, ``after_each``, or ``after_all`` hook.
+                Defaults to ``"failed"``.
         """
         if not self._ensure_client():
             return