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):