From 65a4b7cede2e5aa597aa7b22334ab66e660f646c Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Tue, 11 Aug 2026 15:54:27 -0300 Subject: [PATCH] fix: emit FlowStartedEvent when a boundary hook aborts the flow (#6953) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: emit FlowStartedEvent when a boundary hook aborts the flow A HookAborted at EXECUTION_START or INPUT propagated before `FlowStartedEvent` was emitted, so a policy deny left logs but no record of the execution. On abort, stamp the state id and open the flow scope before re-raising: the deny surfaces as a started -> failed execution while normal runs keep the existing ordering — the started event carries hook-resolved inputs and `id` rewrites keep redirecting persistence restoration. * docs: translate execution-boundary-hooks page to ar, ko, and pt-BR The English page updated on this branch had never been localized. Translate it into the three supported locales following `DOCS_TRANSLATIONS.md` and register the page in each locale's navigation in `docs/docs.json`. Untranslated link targets (the step-hooks page and the aborting-an-operation anchor) are omitted rather than pointed at English, matching the locale navigation convention. --- docs/docs.json | 9 +- .../ar/learn/execution-boundary-hooks.mdx | 200 ++++++++++++++ .../en/learn/execution-boundary-hooks.mdx | 25 +- .../ko/learn/execution-boundary-hooks.mdx | 204 +++++++++++++++ .../pt-BR/learn/execution-boundary-hooks.mdx | 211 +++++++++++++++ .../src/crewai/flow/runtime/__init__.py | 170 ++++++------ .../hooks/test_interception_conformance.py | 245 ++++++++++++++++++ lib/crewai/tests/utilities/test_events.py | 16 +- 8 files changed, 997 insertions(+), 83 deletions(-) create mode 100644 docs/edge/ar/learn/execution-boundary-hooks.mdx create mode 100644 docs/edge/ko/learn/execution-boundary-hooks.mdx create mode 100644 docs/edge/pt-BR/learn/execution-boundary-hooks.mdx diff --git a/docs/docs.json b/docs/docs.json index c348020879..1e5d45ca27 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -12394,7 +12394,8 @@ "edge/pt-BR/learn/using-annotations", "edge/pt-BR/learn/execution-hooks", "edge/pt-BR/learn/llm-hooks", - "edge/pt-BR/learn/tool-hooks" + "edge/pt-BR/learn/tool-hooks", + "edge/pt-BR/learn/execution-boundary-hooks" ] }, { @@ -23684,7 +23685,8 @@ "edge/ko/learn/using-annotations", "edge/ko/learn/execution-hooks", "edge/ko/learn/llm-hooks", - "edge/ko/learn/tool-hooks" + "edge/ko/learn/tool-hooks", + "edge/ko/learn/execution-boundary-hooks" ] }, { @@ -35346,7 +35348,8 @@ "edge/ar/learn/using-annotations", "edge/ar/learn/execution-hooks", "edge/ar/learn/llm-hooks", - "edge/ar/learn/tool-hooks" + "edge/ar/learn/tool-hooks", + "edge/ar/learn/execution-boundary-hooks" ] }, { diff --git a/docs/edge/ar/learn/execution-boundary-hooks.mdx b/docs/edge/ar/learn/execution-boundary-hooks.mdx new file mode 100644 index 0000000000..b660ca3771 --- /dev/null +++ b/docs/edge/ar/learn/execution-boundary-hooks.mdx @@ -0,0 +1,200 @@ +--- +title: خطافات حدود التنفيذ +description: اعتراض بداية تنفيذ الـ Crew والـ Flow ومدخلاته ومخرجاته ونهايته باستخدام المزخرف @on +mode: "wide" +--- + +تعترض خطافات حدود التنفيذ الأطراف الخارجية للتشغيل — قبل بدء أي عمل، وعند +حسم المدخلات، وعند جاهزية النتيجة النهائية، وعند انتهاء التنفيذ. وهي تعمل مع +الـ Crew والـ Flow على حد سواء، وتُعد المكان المناسب لفحوصات السياسة على +مستوى التشغيل وإعادة كتابة المدخلات وتنقية المخرجات. + +## نظرة عامة + +أربع نقاط اعتراض تغطي الحدود: + +| النقطة | التوقيت | `ctx.payload` | +|--------|---------|---------------| +| `EXECUTION_START` | Crew أو Flow على وشك البدء | `dict` المدخلات | +| `INPUT` | المدخلات المحسومة للتنفيذ | `dict` المدخلات | +| `OUTPUT` | النتيجة النهائية جاهزة | كائن المخرجات | +| `EXECUTION_END` | انتهى التنفيذ (نجاحًا أو فشلًا) | كائن المخرجات، أو `None` عند الفشل | + +بالنسبة إلى الـ Crew، يكون payload المخرجات `CrewOutput`. أما في الـ Flow فهو +النتيجة النهائية لدالة الـ Flow. + +## توقيع الخطاف + +```python +from crewai.hooks import on, HookAborted, InterceptionPoint + +@on(InterceptionPoint.EXECUTION_START) +def boundary_hook(ctx) -> Any | None: + # Mutate ctx.payload in place, or + # return a non-None value to replace it, or + # raise HookAborted(reason, source) to stop the run + return None +``` + +تتبع خطافات الحدود العقد القياسي: المتابعة (`return None`)، أو التعديل في +المكان، أو الاستبدال بإرجاع قيمة، أو الإجهاض برفع `HookAborted`. أي إجهاض +عند أي حد ينتشر خارج `kickoff()` مع سببه. + +## مخطط السياق + +تتلقى كل نقطة سياقًا منمّطًا. تشترك جميع السياقات في الحقول الأساسية: + +```python +class InterceptionContext: + payload: Any # The interceptable value (see table above) + agent: Any = None # Not populated at execution boundaries + agent_role: str | None # Not populated at execution boundaries + task: Any = None # Not populated at execution boundaries + crew: Any = None # The Crew instance (crew runs only) + flow: Any = None # The Flow instance (flow runs only) +``` + +تضيف سياقات كل نقطة اسمًا بديلًا للـ payload: + +```python +class ExecutionStartContext(InterceptionContext): + inputs: dict # Same dict as payload + +class InputContext(InterceptionContext): + inputs: dict # Same dict as payload + +class OutputContext(InterceptionContext): + output: Any # The output object + +class ExecutionEndContext(InterceptionContext): + output: Any # The output object (None when status == "failed") + status: str # "completed" or "failed" + error: BaseException | None # The exception when status == "failed" +``` + + +`ctx.inputs` هو اسم بديل لقاموس المدخلات **الأصلي**، لذا فإن التعديلات في +المكان عبر أي من الاسمين متكافئة. إذا *استبدل* خطاف سابق الـ payload بإرجاع +dict جديد، فإن `ctx.payload` وحده يُعاد ربطه — اقرأ واكتب دائمًا عبر +`ctx.payload` عندما يمكن أن تتسلسل الخطافات. + + +## تشغيلات الـ Crew مقابل تشغيلات الـ Flow + +تعمل خطافات الحدود على كلا وقتي التشغيل، وتنفيذ الـ Crew يجري داخليًا فوق وقت +تشغيل Flow. لذلك أثناء `crew.kickoff()` يُطلق الخطاف الحدودي العام لحدّ الـ +Crew (‏`ctx.crew` مضبوط و`ctx.flow` يساوي `None`) **و** للـ Flow الداخلي +(‏`ctx.flow` مضبوط و`ctx.crew` يساوي `None`). ميّز حسب وقت التشغيل: + +```python +@on(InterceptionPoint.OUTPUT) +def crew_output_only(ctx): + if ctx.crew is None: + return None # Skip the internal flow (or a bare flow) + ctx.payload.raw = ctx.payload.raw.strip() +``` + +## حالات استخدام شائعة + +### فحص السياسة عند البدء + +```python +@on(InterceptionPoint.EXECUTION_START) +def enforce_policy(ctx): + if ctx.crew is not None and not ctx.payload.get("authorized"): + raise HookAborted(reason="unauthorized execution", source="access-control") +``` + +### إعادة كتابة المدخلات + +```python +@on(InterceptionPoint.INPUT) +def add_defaults(ctx): + if ctx.crew is None: + return None + ctx.payload.setdefault("locale", "en-US") + ctx.payload["topic"] = ctx.payload["topic"].strip().lower() +``` + +تتدفق المدخلات المعاد كتابتها إلى استيفاء الـ Task، فيتصرف التشغيل كما لو +بدأ بالقاموس المعدل. + +فضّل `INPUT` لإعادة الكتابة وعامل `EXECUTION_START` كبوابة سماح/منع. إعادة +الكتابة عند `EXECUTION_START` تظل مُحترمة — في الـ Crew تغذي أيضًا استدعاءات +`before_kickoff`؛ وفي الـ Flow تُطبق تمامًا كإعادة كتابة `INPUT`. + +### تنقية المخرجات + +```python +import re + +@on(InterceptionPoint.OUTPUT) +def redact_emails(ctx): + if ctx.crew is None: + return None + ctx.payload.raw = re.sub( + r"\b[\w.+-]+@[\w-]+\.[\w.]+\b", "[EMAIL-REDACTED]", ctx.payload.raw + ) +``` + +يعمل `OUTPUT` قبل `EXECUTION_END`، وكلاهما يرى الـ payload (الذي ربما +استُبدل) من الخطافات السابقة؛ والقيمة النهائية المعاد كتابتها هي ما يعيده +`kickoff()`. + +### مراقبة الإخفاقات + +يُطلق `EXECUTION_END` مرة واحدة بالضبط لكل تنفيذ، عند النجاح والفشل على حد +سواء. عندما يرفع التشغيل استثناءً — خطأ في Task، أو استثناء في دالة Flow، أو +`HookAborted` من نقطة سابقة — يتلقى الخطاف `status="failed"` مع الاستثناء في +`ctx.error`، ويظل الاستثناء الأصلي ينتشر خارج `kickoff()` دون تغيير: + +```python +@on(InterceptionPoint.EXECUTION_END) +def report_outcome(ctx): + if ctx.status == "failed": + notify_policy_engine(status="failed", error=repr(ctx.error)) + else: + notify_policy_engine(status="completed") +``` + +تنبيهان: لا يُطلق `EXECUTION_END` عندما لا يكون `EXECUTION_START` قد أُرسل +أصلًا (الإجهاض عند البدء يعني أن الحد لم يُفتح قط، فلا توجد نهاية تقابله)، +ورفع `HookAborted` من إرسال `EXECUTION_END` في مسار الفشل يُتجاهل — لم يعد +هناك ما يُجهض، والخطأ الأصلي هو الغالب. + +## الترتيب + +لتشغيل Crew يكون ترتيب الحدود: + +``` +EXECUTION_START → before_kickoff callbacks → INPUT → tasks execute → OUTPUT → EXECUTION_END +``` + +لتشغيل Flow، تحسم خطافات الحدود المدخلات قبل أن تبدأ أحداث دورة الحياة: + +``` +EXECUTION_START → INPUT → FlowStartedEvent → flow methods execute → OUTPUT → EXECUTION_END → FlowFinishedEvent +``` + +يحمل `FlowStartedEvent` المدخلات كما حسمتها الخطافات، وإعادة كتابة +`inputs["id"]` داخل خطاف حدودي تعيد توجيه استعادة الحالة. يظهر الإجهاض عند +`EXECUTION_START` مع ذلك كحدث `FlowStartedEvent` يتبعه `FlowFailedEvent`، +ويُبثان عند الإجهاض مع الحمولة كما حسمتها الخطافات التي عملت قبله. + +تعمل الخطافات في النقطة نفسها حسب ترتيب التسجيل، الخطافات العامة أولًا ثم +الخطافات المحدودة بالـ Crew. تُبث القياسات (`HookDispatchedEvent`) مع كل +إرسال. + +## إدارة الخطافات في الاختبارات + +```python +from crewai.hooks import clear_all_hooks + +clear_all_hooks() # Clears every point, including boundaries +``` + +## وثائق ذات صلة + +- [نظرة عامة على خطافات التنفيذ →](/edge/ar/learn/execution-hooks) +- [خطافات استدعاء LLM →](/edge/ar/learn/llm-hooks) +- [خطافات استدعاء الأدوات →](/edge/ar/learn/tool-hooks) diff --git a/docs/edge/en/learn/execution-boundary-hooks.mdx b/docs/edge/en/learn/execution-boundary-hooks.mdx index 80e2b382e0..f1841f4041 100644 --- a/docs/edge/en/learn/execution-boundary-hooks.mdx +++ b/docs/edge/en/learn/execution-boundary-hooks.mdx @@ -120,6 +120,11 @@ def add_defaults(ctx): Rewritten inputs flow into task interpolation, so the run behaves as if it was kicked off with the modified dict. +Prefer `INPUT` for rewriting and treat `EXECUTION_START` as the allow/deny +gate. Rewrites at `EXECUTION_START` are still honored — on crews they also +feed the `before_kickoff` callbacks; on flows they land exactly like an +`INPUT` rewrite. + ### Output Sanitization ```python @@ -156,9 +161,10 @@ def report_outcome(ctx): ``` Two caveats: `EXECUTION_END` does not fire when `EXECUTION_START` never -dispatched (an abort at start counts as the execution never beginning), and -raising `HookAborted` from a failure-path `EXECUTION_END` dispatch is ignored — -there is nothing left to abort, and the original error wins. +dispatched (an abort at start means the boundary never opened, so there is no +end to pair), and raising `HookAborted` from a failure-path `EXECUTION_END` +dispatch is ignored — there is nothing left to abort, and the original error +wins. ## Ordering @@ -168,6 +174,19 @@ For a crew run the boundary order is: EXECUTION_START → before_kickoff callbacks → INPUT → tasks execute → OUTPUT → EXECUTION_END ``` +For a flow run, the boundary hooks resolve the inputs before the lifecycle +events begin: + +``` +EXECUTION_START → INPUT → FlowStartedEvent → flow methods execute → OUTPUT → EXECUTION_END → FlowFinishedEvent +``` + +`FlowStartedEvent` carries the hook-resolved inputs, and rewriting +`inputs["id"]` in a boundary hook redirects state restoration. An abort at +`EXECUTION_START` still surfaces as `FlowStartedEvent` followed by +`FlowFailedEvent`, emitted at the abort with the payload as resolved by the +hooks that ran before it. + Hooks at the same point run in registration order, global hooks first, then crew-scoped hooks. Telemetry (`HookDispatchedEvent`) is emitted per dispatch. diff --git a/docs/edge/ko/learn/execution-boundary-hooks.mdx b/docs/edge/ko/learn/execution-boundary-hooks.mdx new file mode 100644 index 0000000000..c68a776000 --- /dev/null +++ b/docs/edge/ko/learn/execution-boundary-hooks.mdx @@ -0,0 +1,204 @@ +--- +title: 실행 경계 훅 +description: "@on 데코레이터로 crew와 flow 실행의 시작, 입력, 출력, 종료를 가로채기" +mode: "wide" +--- + +실행 경계 훅은 실행의 가장 바깥쪽 경계를 가로챕니다 — 작업이 시작되기 전, +입력이 확정될 때, 최종 결과가 준비될 때, 그리고 실행이 끝날 때입니다. 크루와 +플로우 모두에서 발생하며, 실행 수준의 정책 검사, 입력 재작성, 출력 정제에 +적합한 위치입니다. + +## 개요 + +네 가지 인터셉션 포인트가 경계를 담당합니다: + +| 포인트 | 시점 | `ctx.payload` | +|--------|------|---------------| +| `EXECUTION_START` | 크루 또는 플로우가 막 시작되려는 시점 | 입력 `dict` | +| `INPUT` | 실행을 위한 입력이 확정된 시점 | 입력 `dict` | +| `OUTPUT` | 최종 결과가 준비된 시점 | 출력 객체 | +| `EXECUTION_END` | 실행이 끝난 시점(성공 또는 실패) | 출력 객체, 실패 시 `None` | + +크루의 경우 출력 payload는 `CrewOutput`입니다. 플로우의 경우 최종 플로우 +메서드의 결과입니다. + +## 훅 시그니처 + +```python +from crewai.hooks import on, HookAborted, InterceptionPoint + +@on(InterceptionPoint.EXECUTION_START) +def boundary_hook(ctx) -> Any | None: + # Mutate ctx.payload in place, or + # return a non-None value to replace it, or + # raise HookAborted(reason, source) to stop the run + return None +``` + +경계 훅은 표준 계약을 따릅니다: 진행(`return None`), 제자리(in-place) 수정, +값을 반환하여 교체, 또는 `HookAborted`를 발생시켜 중단합니다. 어떤 +경계에서든 중단(abort)은 그 사유와 함께 `kickoff()` 밖으로 전파됩니다. + +## 컨텍스트 스키마 + +각 포인트는 타입이 지정된 컨텍스트를 받습니다. 모든 컨텍스트는 공통 기본 +필드를 공유합니다: + +```python +class InterceptionContext: + payload: Any # The interceptable value (see table above) + agent: Any = None # Not populated at execution boundaries + agent_role: str | None # Not populated at execution boundaries + task: Any = None # Not populated at execution boundaries + crew: Any = None # The Crew instance (crew runs only) + flow: Any = None # The Flow instance (flow runs only) +``` + +포인트별 컨텍스트는 payload에 대한 이름 있는 별칭을 추가합니다: + +```python +class ExecutionStartContext(InterceptionContext): + inputs: dict # Same dict as payload + +class InputContext(InterceptionContext): + inputs: dict # Same dict as payload + +class OutputContext(InterceptionContext): + output: Any # The output object + +class ExecutionEndContext(InterceptionContext): + output: Any # The output object (None when status == "failed") + status: str # "completed" or "failed" + error: BaseException | None # The exception when status == "failed" +``` + + +`ctx.inputs`는 **원본** 입력 dict의 별칭이므로, 어느 이름으로든 제자리 +수정은 동일하게 동작합니다. 이전 훅이 새 dict를 반환하여 payload를 +*교체*했다면 `ctx.payload`만 다시 바인딩됩니다 — 훅이 연쇄될 수 있는 경우 +항상 `ctx.payload`를 읽고 쓰세요. + + +## 크루 실행 vs. 플로우 실행 + +경계 훅은 두 런타임 모두에서 발생하며, 크루 실행은 내부적으로 플로우 런타임 +위에서 동작합니다. 따라서 `crew.kickoff()` 중에는 전역 경계 훅이 크루 +경계(`ctx.crew` 설정, `ctx.flow`는 `None`)**와** 내부 플로우(`ctx.flow` +설정, `ctx.crew`는 `None`) 모두에서 발생합니다. 런타임으로 구분하세요: + +```python +@on(InterceptionPoint.OUTPUT) +def crew_output_only(ctx): + if ctx.crew is None: + return None # Skip the internal flow (or a bare flow) + ctx.payload.raw = ctx.payload.raw.strip() +``` + +## 일반적인 사용 사례 + +### 시작 시 정책 검사 + +```python +@on(InterceptionPoint.EXECUTION_START) +def enforce_policy(ctx): + if ctx.crew is not None and not ctx.payload.get("authorized"): + raise HookAborted(reason="unauthorized execution", source="access-control") +``` + +### 입력 재작성 + +```python +@on(InterceptionPoint.INPUT) +def add_defaults(ctx): + if ctx.crew is None: + return None + ctx.payload.setdefault("locale", "en-US") + ctx.payload["topic"] = ctx.payload["topic"].strip().lower() +``` + +재작성된 입력은 태스크 보간(interpolation)으로 흘러가므로, 실행은 수정된 +dict로 시작된 것처럼 동작합니다. + +재작성에는 `INPUT`을 사용하고, `EXECUTION_START`는 허용/거부 게이트로 +취급하세요. `EXECUTION_START`에서의 재작성도 여전히 반영됩니다 — 크루에서는 +`before_kickoff` 콜백에도 전달되고, 플로우에서는 `INPUT` 재작성과 동일하게 +적용됩니다. + +### 출력 정제 + +```python +import re + +@on(InterceptionPoint.OUTPUT) +def redact_emails(ctx): + if ctx.crew is None: + return None + ctx.payload.raw = re.sub( + r"\b[\w.+-]+@[\w-]+\.[\w.]+\b", "[EMAIL-REDACTED]", ctx.payload.raw + ) +``` + +`OUTPUT`은 `EXECUTION_END`보다 먼저 실행되며, 둘 다 이전 훅에서 (교체되었을 +수 있는) payload를 봅니다. 최종적으로 재작성된 값이 `kickoff()`가 반환하는 +값입니다. + +### 실패 관찰 + +`EXECUTION_END`는 성공이든 실패든 실행마다 정확히 한 번 발생합니다. 실행이 +예외를 던지면 — 태스크 오류, 플로우 메서드 예외, 또는 이전 포인트의 +`HookAborted` — 훅은 `ctx.error`에 예외가 담긴 `status="failed"`를 받으며, +원래 예외는 변경 없이 `kickoff()` 밖으로 전파됩니다: + +```python +@on(InterceptionPoint.EXECUTION_END) +def report_outcome(ctx): + if ctx.status == "failed": + notify_policy_engine(status="failed", error=repr(ctx.error)) + else: + notify_policy_engine(status="completed") +``` + +두 가지 주의 사항: `EXECUTION_START`가 디스패치되지 않았다면 +`EXECUTION_END`는 발생하지 않습니다(시작 시점의 중단은 경계가 열리지 +않았다는 뜻이므로 짝을 이룰 종료가 없습니다). 또한 실패 경로의 +`EXECUTION_END` 디스패치에서 `HookAborted`를 발생시키는 것은 무시됩니다 — +더 이상 중단할 것이 없고, 원래 오류가 우선합니다. + +## 순서 + +크루 실행의 경계 순서는 다음과 같습니다: + +``` +EXECUTION_START → before_kickoff callbacks → INPUT → tasks execute → OUTPUT → EXECUTION_END +``` + +플로우 실행에서는 라이프사이클 이벤트가 시작되기 전에 경계 훅이 입력을 +확정합니다: + +``` +EXECUTION_START → INPUT → FlowStartedEvent → flow methods execute → OUTPUT → EXECUTION_END → FlowFinishedEvent +``` + +`FlowStartedEvent`는 훅이 확정한 입력을 담으며, 경계 훅에서 `inputs["id"]`를 +재작성하면 상태 복원 대상이 바뀝니다. `EXECUTION_START`에서의 중단은 여전히 +`FlowStartedEvent` 다음에 `FlowFailedEvent`가 오는 형태로 나타나며, 중단 +시점에 그때까지 실행된 훅이 확정한 페이로드와 함께 발생합니다. + +같은 포인트의 훅은 등록 순서대로 실행되며, 전역 훅이 먼저, 그다음 크루 범위 +훅이 실행됩니다. 텔레메트리(`HookDispatchedEvent`)는 디스패치마다 +발생합니다. + +## 테스트에서 훅 관리 + +```python +from crewai.hooks import clear_all_hooks + +clear_all_hooks() # Clears every point, including boundaries +``` + +## 관련 문서 + +- [실행 훅 개요 →](/edge/ko/learn/execution-hooks) +- [LLM 호출 훅 →](/edge/ko/learn/llm-hooks) +- [도구 호출 훅 →](/edge/ko/learn/tool-hooks) diff --git a/docs/edge/pt-BR/learn/execution-boundary-hooks.mdx b/docs/edge/pt-BR/learn/execution-boundary-hooks.mdx new file mode 100644 index 0000000000..067e531637 --- /dev/null +++ b/docs/edge/pt-BR/learn/execution-boundary-hooks.mdx @@ -0,0 +1,211 @@ +--- +title: Hooks de Fronteira de Execução +description: Intercepte o início, as entradas, a saída e o fim de execuções de crews e flows com o decorator @on +mode: "wide" +--- + +Os hooks de fronteira de execução interceptam as bordas mais externas de uma +execução — antes de qualquer trabalho começar, quando as entradas são +resolvidas, quando o resultado final está pronto e quando a execução termina. +Eles disparam tanto para crews quanto para flows e são o lugar certo para +verificações de política no nível da execução, reescrita de entradas e +sanitização de saídas. + +## Visão Geral + +Quatro pontos de interceptação cobrem as fronteiras: + +| Ponto | Quando | `ctx.payload` | +|-------|--------|---------------| +| `EXECUTION_START` | Uma crew ou flow está prestes a começar | `dict` de entradas | +| `INPUT` | Entradas resolvidas para a execução | `dict` de entradas | +| `OUTPUT` | O resultado final está pronto | o objeto de saída | +| `EXECUTION_END` | A execução terminou (sucesso ou falha) | o objeto de saída, ou `None` em caso de falha | + +Para uma crew, o payload de saída é um `CrewOutput`. Para um flow, é o +resultado final do método do flow. + +## Assinatura do Hook + +```python +from crewai.hooks import on, HookAborted, InterceptionPoint + +@on(InterceptionPoint.EXECUTION_START) +def boundary_hook(ctx) -> Any | None: + # Mutate ctx.payload in place, or + # return a non-None value to replace it, or + # raise HookAborted(reason, source) to stop the run + return None +``` + +Hooks de fronteira seguem o contrato padrão: prosseguir (`return None`), mutar +in place, substituir retornando um valor, ou abortar lançando `HookAborted`. +Um abort em qualquer fronteira propaga para fora do `kickoff()` com seu +motivo. + +## Esquema de Contexto + +Cada ponto recebe um contexto tipado. Todos os contextos compartilham os +campos base: + +```python +class InterceptionContext: + payload: Any # The interceptable value (see table above) + agent: Any = None # Not populated at execution boundaries + agent_role: str | None # Not populated at execution boundaries + task: Any = None # Not populated at execution boundaries + crew: Any = None # The Crew instance (crew runs only) + flow: Any = None # The Flow instance (flow runs only) +``` + +Os contextos de cada ponto adicionam um alias nomeado para o payload: + +```python +class ExecutionStartContext(InterceptionContext): + inputs: dict # Same dict as payload + +class InputContext(InterceptionContext): + inputs: dict # Same dict as payload + +class OutputContext(InterceptionContext): + output: Any # The output object + +class ExecutionEndContext(InterceptionContext): + output: Any # The output object (None when status == "failed") + status: str # "completed" or "failed" + error: BaseException | None # The exception when status == "failed" +``` + + +`ctx.inputs` é um alias para o dict de entradas **original**, então edições in +place por qualquer um dos nomes são equivalentes. Se um hook anterior +*substituiu* o payload retornando um novo dict, apenas `ctx.payload` é +reassociado — sempre leia e escreva `ctx.payload` quando hooks puderem +encadear. + + +## Execuções de Crew vs. Execuções de Flow + +Hooks de fronteira disparam em ambos os runtimes, e a execução de uma crew +roda internamente sobre um runtime de flow. Durante um `crew.kickoff()`, um +hook de fronteira global portanto dispara para a fronteira da crew +(`ctx.crew` definido, `ctx.flow` `None`) **e** para o flow interno +(`ctx.flow` definido, `ctx.crew` `None`). Discrimine pelo runtime: + +```python +@on(InterceptionPoint.OUTPUT) +def crew_output_only(ctx): + if ctx.crew is None: + return None # Skip the internal flow (or a bare flow) + ctx.payload.raw = ctx.payload.raw.strip() +``` + +## Casos de Uso Comuns + +### Verificação de Política no Início + +```python +@on(InterceptionPoint.EXECUTION_START) +def enforce_policy(ctx): + if ctx.crew is not None and not ctx.payload.get("authorized"): + raise HookAborted(reason="unauthorized execution", source="access-control") +``` + +### Reescrita de Entradas + +```python +@on(InterceptionPoint.INPUT) +def add_defaults(ctx): + if ctx.crew is None: + return None + ctx.payload.setdefault("locale", "en-US") + ctx.payload["topic"] = ctx.payload["topic"].strip().lower() +``` + +Entradas reescritas fluem para a interpolação de tasks, então a execução se +comporta como se tivesse sido iniciada com o dict modificado. + +Prefira `INPUT` para reescrita e trate `EXECUTION_START` como o gate de +allow/deny. Reescritas em `EXECUTION_START` continuam sendo honradas — em +crews elas também alimentam os callbacks de `before_kickoff`; em flows elas +se aplicam exatamente como uma reescrita de `INPUT`. + +### Sanitização de Saída + +```python +import re + +@on(InterceptionPoint.OUTPUT) +def redact_emails(ctx): + if ctx.crew is None: + return None + ctx.payload.raw = re.sub( + r"\b[\w.+-]+@[\w-]+\.[\w.]+\b", "[EMAIL-REDACTED]", ctx.payload.raw + ) +``` + +`OUTPUT` roda antes de `EXECUTION_END`, e ambos veem o payload (possivelmente +substituído) de hooks anteriores; o valor final reescrito é o que `kickoff()` +retorna. + +### Observando Falhas + +`EXECUTION_END` dispara exatamente uma vez por execução, tanto em sucesso +quanto em falha. Quando a execução lança uma exceção — um erro de task, uma +exceção de método de flow ou um `HookAborted` de um ponto anterior — o hook +recebe `status="failed"` com a exceção em `ctx.error`, e a exceção original +ainda propaga para fora do `kickoff()` sem alterações: + +```python +@on(InterceptionPoint.EXECUTION_END) +def report_outcome(ctx): + if ctx.status == "failed": + notify_policy_engine(status="failed", error=repr(ctx.error)) + else: + notify_policy_engine(status="completed") +``` + +Duas ressalvas: `EXECUTION_END` não dispara quando `EXECUTION_START` nunca foi +despachado (um abort no início significa que a fronteira nunca abriu, então +não há fim para parear), e lançar `HookAborted` de um dispatch de +`EXECUTION_END` no caminho de falha é ignorado — não resta nada para abortar, +e o erro original prevalece. + +## Ordenação + +Para uma execução de crew, a ordem de fronteira é: + +``` +EXECUTION_START → before_kickoff callbacks → INPUT → tasks execute → OUTPUT → EXECUTION_END +``` + +Para uma execução de flow, os hooks de fronteira resolvem as entradas antes +de os eventos de ciclo de vida começarem: + +``` +EXECUTION_START → INPUT → FlowStartedEvent → flow methods execute → OUTPUT → EXECUTION_END → FlowFinishedEvent +``` + +`FlowStartedEvent` carrega as entradas resolvidas pelos hooks, e reescrever +`inputs["id"]` em um hook de fronteira redireciona a restauração de estado. +Um abort em `EXECUTION_START` ainda aparece como `FlowStartedEvent` seguido +de `FlowFailedEvent`, emitidos no momento do abort com o payload como +resolvido pelos hooks que rodaram antes dele. + +Hooks no mesmo ponto rodam em ordem de registro, hooks globais primeiro, +depois hooks com escopo de crew. A telemetria (`HookDispatchedEvent`) é +emitida por dispatch. + +## Gerenciando Hooks em Testes + +```python +from crewai.hooks import clear_all_hooks + +clear_all_hooks() # Clears every point, including boundaries +``` + +## Documentação Relacionada + +- [Visão Geral dos Hooks de Execução →](/edge/pt-BR/learn/execution-hooks) +- [Hooks de Chamada LLM →](/edge/pt-BR/learn/llm-hooks) +- [Hooks de Chamada de Ferramenta →](/edge/pt-BR/learn/tool-hooks) diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py index 4bb78f9fd5..46278f4fe1 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/lib/crewai/src/crewai/flow/runtime/__init__.py @@ -2142,28 +2142,42 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): ExecutionEndContext, ExecutionStartContext, InputContext, + InterceptionContext, OutputContext, ) - from crewai.hooks.dispatch import InterceptionPoint, dispatch + from crewai.hooks.dispatch import HookAborted, InterceptionPoint, dispatch # ``inputs`` aliases the same object as ``payload`` (not a fresh # ``{}`` from ``or``) so in-place edits survive read-back. - start_ctx = ExecutionStartContext( - flow=self, - inputs=inputs if inputs is not None else {}, - payload=inputs, - ) - dispatch(InterceptionPoint.EXECUTION_START, start_ctx) - execution_start_dispatched = True - inputs = start_ctx.payload + try: + boundary_ctx: InterceptionContext = ExecutionStartContext( + flow=self, + inputs=inputs if inputs is not None else {}, + payload=inputs, + ) + dispatch(InterceptionPoint.EXECUTION_START, boundary_ctx) + execution_start_dispatched = True + inputs = boundary_ctx.payload - input_ctx = InputContext( - flow=self, - inputs=inputs if inputs is not None else {}, - payload=inputs, - ) - dispatch(InterceptionPoint.INPUT, input_ctx) - inputs = input_ctx.payload + boundary_ctx = InputContext( + flow=self, + inputs=inputs if inputs is not None else {}, + payload=inputs, + ) + dispatch(InterceptionPoint.INPUT, boundary_ctx) + inputs = boundary_ctx.payload + except HookAborted: + # The deny surfaces as started -> failed. Read the payload back + # from the aborted dispatch first: earlier hooks in the chain + # may have replaced it before a later one aborted. Then stamp + # the state id so failure listeners correlate the record, open + # the flow scope, and re-raise so the failure pairs with the + # opener. + inputs = boundary_ctx.payload + if inputs and "id" in inputs: + self._stamp_state_id(inputs["id"]) + flow_scope_open = await self._open_flow_scope(inputs) + raise # Publish the resolved inputs so trigger-payload injection and other # baggage readers observe hook rewrites (the baggage set before the @@ -2214,10 +2228,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): new_state_id = (inputs.get("id") if inputs else None) or str( uuid4() ) - if isinstance(self._state, dict): - self._state["id"] = new_state_id - elif isinstance(self._state, BaseModel): - setattr(self._state, "id", new_state_id) # noqa: B010 + self._stamp_state_id(new_state_id) fork_succeeded = True else: self._log_flow_event( @@ -2230,10 +2241,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): # Override the id in the state if it exists in inputs. # Skip when the fork already assigned state.id above. if "id" in inputs and not fork_succeeded: - if isinstance(self._state, dict): - self._state["id"] = inputs["id"] - elif isinstance(self._state, BaseModel): - setattr(self._state, "id", inputs["id"]) # noqa: B010 + self._stamp_state_id(inputs["id"]) # If persistence is enabled, attempt to restore the stored state using the provided id. # Skip when the fork already restored self._state above. @@ -2251,7 +2259,8 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): self._restore_state(stored_state) else: self._log_flow_event( - f"No flow state found for UUID: {restore_uuid}", color="red" + f"No flow state found for UUID: {restore_uuid}", + color="red", ) # Update state with any additional inputs (ignoring the 'id' key) @@ -2259,55 +2268,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): if filtered_inputs: self._initialize_state(filtered_inputs) - defer_trace_finalization = self._should_defer_trace_finalization() - deferred_started_event_id = self._deferred_flow_started_event_id - should_emit_flow_started = not ( - defer_trace_finalization and deferred_started_event_id - ) - if current_flow_id.get() == self.flow_id: - TraceCollectionListener().batch_manager.defer_session_finalization = ( - defer_trace_finalization - ) - - if ( - defer_trace_finalization - and deferred_started_event_id - and get_current_parent_id() is None - ): - restore_event_scope(((deferred_started_event_id, "flow_started"),)) - flow_scope_open = True - elif get_current_parent_id() is None: - reset_emission_counter() - reset_last_event_id() - - if should_emit_flow_started: - # In normal flows, each kickoff owns its own flow lifecycle. - # Deferred sessions reuse the first flow scope until an - # explicit finalization call closes the batch. - started_event = FlowStartedEvent( - type="flow_started", - flow_name=self._definition.name, - inputs=inputs, - ) - future = crewai_event_bus.emit(self, started_event) - flow_scope_open = True - if future: - try: - await asyncio.wrap_future(future) - except Exception: - logger.warning("FlowStartedEvent handler failed", exc_info=True) - # Stash the started event id so a deferred - # ``finalize_session_traces()`` can restore the event scope - # before emitting ``FlowFinishedEvent`` (otherwise the bus - # warns "Ending event 'flow_finished' emitted with empty - # scope stack"). - if defer_trace_finalization: - object.__setattr__( - self, "_deferred_flow_started_event_id", started_event.event_id - ) - # After FlowStarted: env events must not pre-empt trace batch init - # with implicit "crew" execution_type. - get_env_context() + flow_scope_open = await self._open_flow_scope(inputs) if self._should_apply_pending_kickoff_context(): self._apply_pending_kickoff_context() @@ -2531,6 +2492,67 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): except Exception: # noqa: S110 - aborting an already-failed execution is meaningless pass + def _stamp_state_id(self, state_id: str) -> None: + if isinstance(self._state, dict): + self._state["id"] = state_id + elif isinstance(self._state, BaseModel): + setattr(self._state, "id", state_id) # noqa: B010 + + async def _open_flow_scope(self, inputs: dict[str, Any] | None) -> bool: + """Emit ``FlowStartedEvent`` (or restore a deferred session's event + scope) and return whether the flow scope is open.""" + defer_trace_finalization = self._should_defer_trace_finalization() + deferred_started_event_id = self._deferred_flow_started_event_id + should_emit_flow_started = not ( + defer_trace_finalization and deferred_started_event_id + ) + if current_flow_id.get() == self.flow_id: + TraceCollectionListener().batch_manager.defer_session_finalization = ( + defer_trace_finalization + ) + + flow_scope_open = False + if ( + defer_trace_finalization + and deferred_started_event_id + and get_current_parent_id() is None + ): + restore_event_scope(((deferred_started_event_id, "flow_started"),)) + flow_scope_open = True + elif get_current_parent_id() is None: + reset_emission_counter() + reset_last_event_id() + + if should_emit_flow_started: + # In normal flows, each kickoff owns its own flow lifecycle. + # Deferred sessions reuse the first flow scope until an + # explicit finalization call closes the batch. + started_event = FlowStartedEvent( + type="flow_started", + flow_name=self._definition.name, + inputs=inputs, + ) + future = crewai_event_bus.emit(self, started_event) + flow_scope_open = True + if future: + try: + await asyncio.wrap_future(future) + except Exception: + logger.warning("FlowStartedEvent handler failed", exc_info=True) + # Stash the started event id so a deferred + # ``finalize_session_traces()`` can restore the event scope + # before emitting ``FlowFinishedEvent`` (otherwise the bus + # warns "Ending event 'flow_finished' emitted with empty + # scope stack"). + if defer_trace_finalization: + object.__setattr__( + self, "_deferred_flow_started_event_id", started_event.event_id + ) + # After FlowStarted: env events must not pre-empt trace batch init + # with implicit "crew" execution_type. + get_env_context() + return flow_scope_open + async def _emit_flow_failed( self, error: Exception, *, respect_suppression: bool = False ) -> None: diff --git a/lib/crewai/tests/hooks/test_interception_conformance.py b/lib/crewai/tests/hooks/test_interception_conformance.py index c2caff5319..f171c38015 100644 --- a/lib/crewai/tests/hooks/test_interception_conformance.py +++ b/lib/crewai/tests/hooks/test_interception_conformance.py @@ -14,6 +14,11 @@ from crewai.agent import Agent from crewai.crew import Crew from crewai.events.event_bus import crewai_event_bus from crewai.events.types.crew_events import CrewKickoffCompletedEvent +from crewai.events.types.flow_events import ( + FlowFailedEvent, + FlowFinishedEvent, + FlowStartedEvent, +) from crewai.flow.flow import Flow, listen, start from crewai.hooks.dispatch import ( HookAborted, @@ -104,6 +109,18 @@ class TestFlowExecutionBoundaries: _SimpleFlow().kickoff(inputs={"seed": 42}) assert seen == {"seed": 42} + def test_input_hook_rewrite_lands_in_flow_state(self): + @on(InterceptionPoint.INPUT) + def inject(ctx): + return {**(ctx.payload or {}), "injected": "by-hook"} + + class _StateReader(Flow): + @start() + def begin(self): + return self.state["injected"] + + assert _StateReader().kickoff(inputs={"seed": 1}) == "by-hook" + def test_abort_at_execution_start_interrupts(self): @on(InterceptionPoint.EXECUTION_START) def block(ctx): @@ -351,3 +368,231 @@ class TestCrewOutput: assert result.raw.endswith("changed by hook") assert completed_raw assert completed_raw[-1].endswith("changed by hook") + + +class TestFlowLifecycleEventOrdering: + """Boundary hooks resolve the inputs before FlowStartedEvent snapshots + them; an abort at EXECUTION_START still surfaces as started -> failed.""" + + def test_abort_at_execution_start_emits_started_then_failed(self): + lifecycle: list[str] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(FlowStartedEvent) + def on_started(_source, _event): + lifecycle.append("flow_started") + + @crewai_event_bus.on(FlowFailedEvent) + def on_failed(_source, _event): + lifecycle.append("flow_failed") + + @crewai_event_bus.on(FlowFinishedEvent) + def on_finished(_source, _event): + lifecycle.append("flow_finished") + + @on(InterceptionPoint.EXECUTION_START) + def block(ctx): + raise HookAborted(reason="denied", source="policy") + + with pytest.raises(HookAborted, match="denied"): + _SimpleFlow().kickoff() + crewai_event_bus.flush() + + assert lifecycle == ["flow_started", "flow_failed"] + + def test_boundary_id_rewrite_redirects_state_identity(self): + # Listeners key execution records off flow_id at emission time. + seen_ids: list[str] = [] + rewritten_id = "0b8ee866-77b3-4dc6-9de4-90a92cbc9fcf" + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(FlowStartedEvent) + def capture(source, _event): + seen_ids.append(source.flow_id) + + @on(InterceptionPoint.EXECUTION_START) + def rewrite(ctx): + return {**(ctx.payload or {}), "id": rewritten_id} + + _SimpleFlow().kickoff( + inputs={"id": "2f9deb0a-41c5-4f4e-9be9-3a5a34c4dc94"} + ) + crewai_event_bus.flush() + + assert seen_ids == [rewritten_id] + + def test_abort_keeps_state_id_from_inputs(self): + # Error-by-id correlation reads the state id stamped at abort time. + seen_ids: list[str] = [] + pinned_id = "0b8ee866-77b3-4dc6-9de4-90a92cbc9fcf" + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(FlowStartedEvent) + def capture(source, _event): + seen_ids.append(source.flow_id) + + @on(InterceptionPoint.EXECUTION_START) + def block(ctx): + raise HookAborted(reason="denied") + + with pytest.raises(HookAborted): + _SimpleFlow().kickoff(inputs={"id": pinned_id}) + crewai_event_bus.flush() + + assert seen_ids == [pinned_id] + + def test_input_redaction_shields_started_event_inputs(self): + captured: list[dict | None] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(FlowStartedEvent) + def capture(_source, event): + captured.append(event.inputs) + + @on(InterceptionPoint.INPUT) + def redact(ctx): + return {**(ctx.payload or {}), "api_key": "[REDACTED]"} + + _SimpleFlow().kickoff(inputs={"api_key": "sk-live-secret"}) + crewai_event_bus.flush() + + assert captured == [{"api_key": "[REDACTED]"}] + + def test_abort_preserves_chained_hook_rewrite(self): + # A replacement returned by an earlier hook lands on ctx.payload as + # each hook runs; a later abort must not fall back to pre-hook inputs. + seen_ids: list[str] = [] + captured: list[dict | None] = [] + rewritten_id = "0b8ee866-77b3-4dc6-9de4-90a92cbc9fcf" + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(FlowStartedEvent) + def capture(source, event): + seen_ids.append(source.flow_id) + captured.append(event.inputs) + + @on(InterceptionPoint.EXECUTION_START) + def rewrite(ctx): + return {**(ctx.payload or {}), "id": rewritten_id} + + @on(InterceptionPoint.EXECUTION_START) + def block(_ctx): + raise HookAborted(reason="denied") + + with pytest.raises(HookAborted): + _SimpleFlow().kickoff( + inputs={"id": "2f9deb0a-41c5-4f4e-9be9-3a5a34c4dc94"} + ) + crewai_event_bus.flush() + + assert seen_ids == [rewritten_id] + assert captured == [{"id": rewritten_id}] + + def test_abort_at_input_preserves_execution_start_rewrite(self): + captured: list[dict | None] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(FlowStartedEvent) + def capture(_source, event): + captured.append(event.inputs) + + @on(InterceptionPoint.EXECUTION_START) + def redact(ctx): + return {**(ctx.payload or {}), "api_key": "[REDACTED]"} + + @on(InterceptionPoint.INPUT) + def block(_ctx): + raise HookAborted(reason="denied") + + with pytest.raises(HookAborted): + _SimpleFlow().kickoff(inputs={"api_key": "sk-live-secret"}) + crewai_event_bus.flush() + + assert captured == [{"api_key": "[REDACTED]"}] + + def test_kickoff_emits_started_exactly_once(self): + started: list[str] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(FlowStartedEvent) + def capture(_source, event): + started.append(event.flow_name) + + _SimpleFlow().kickoff() + crewai_event_bus.flush() + + assert started == ["_SimpleFlow"] + + def test_reentrant_kickoff_emits_started_once_per_invocation(self): + started: list[str] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(FlowStartedEvent) + def capture(_source, event): + started.append(event.flow_name) + + with pytest.raises(RuntimeError, match="outer boom"): + _ReentrantFailingFlow().kickoff() + crewai_event_bus.flush() + + # One started per kickoff invocation: outer run + its nested kickoff. + assert started == ["_ReentrantFailingFlow", "_ReentrantFailingFlow"] + + def test_deferred_session_deny_at_execution_start_keeps_session_open(self): + started: list[FlowStartedEvent] = [] + failed: list[FlowFailedEvent] = [] + finished: list[FlowFinishedEvent] = [] + + class _DeferredFlow(Flow): + defer_trace_finalization = True + + @start() + def begin(self): + return "turn-result" + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(FlowStartedEvent) + def on_started(_source, event): + started.append(event) + + @crewai_event_bus.on(FlowFailedEvent) + def on_failed(_source, event): + failed.append(event) + + @crewai_event_bus.on(FlowFinishedEvent) + def on_finished(_source, event): + finished.append(event) + + @on(InterceptionPoint.EXECUTION_START) + def block(ctx): + raise HookAborted(reason="denied") + + flow = _DeferredFlow() + with pytest.raises(HookAborted): + flow.kickoff() + crewai_event_bus.flush() + + # Deferred sessions suppress per-turn terminal events; the + # session stays open under the already-emitted started event. + assert len(started) == 1 + assert failed == [] + assert finished == [] + + clear_all() + assert flow.kickoff() == "turn-result" + flow.finalize_session_traces() + crewai_event_bus.flush() + + assert len(started) == 1 + assert failed == [] + assert len(finished) == 1 + assert finished[0].started_event_id == started[0].event_id diff --git a/lib/crewai/tests/utilities/test_events.py b/lib/crewai/tests/utilities/test_events.py index a5e22fe81a..6f2926b227 100644 --- a/lib/crewai/tests/utilities/test_events.py +++ b/lib/crewai/tests/utilities/test_events.py @@ -628,9 +628,10 @@ def test_suppressed_flow_failure_matches_finished_event_emission(): assert len(failed) == 1 -def test_abort_before_flow_started_emits_no_failed_event(): +def test_abort_at_execution_start_emits_started_then_failed_events(): started: list[FlowStartedEvent] = [] failed: list[FlowFailedEvent] = [] + finished: list[FlowFinishedEvent] = [] class BlockedFlow(Flow): @start() @@ -654,14 +655,23 @@ def test_abort_before_flow_started_emits_no_failed_event(): def handle_flow_failed(source, event): failed.append(event) + @crewai_event_bus.on(FlowFinishedEvent) + def handle_flow_finished(source, event): + finished.append(event) + with pytest.raises(HookAborted): BlockedFlow().kickoff() wait_for_event_handlers() finally: clear_all() - assert started == [] - assert failed == [] + assert len(started) == 1 + assert len(failed) == 1 + assert finished == [] + assert failed[0].flow_name == "BlockedFlow" + assert isinstance(failed[0].error, HookAborted) + assert failed[0].error.reason == "blocked by policy" + assert failed[0].started_event_id == started[0].event_id def test_resume_emits_failed_event_paired_with_resume_started_event(tmp_path):