feat(flow): add experimental job lifecycle and runner (#7864)

docs/edge/ar/guides/flows/conversational-flows.mdx

@@ -616,3 +616,78 @@ from crewai.flow import (
 
 - [إتقان إدارة حالة Flow](/ar/guides/flows/mastering-flow-state)
 - [أنشئ أول Flow](/ar/guides/flows/first-flow)
+
+
+## Jobs تجريبية تمتد عبر أدوار المحادثة
+
+هذه API تجريبية وقد تتغير. استوردها صراحةً من `crewai.experimental.flow_jobs`؛ وليست جزءًا من واجهة `crewai.flow` المستقرة.
+
+توفر `JobRecord` و`JobState` و`JobUpdate` و`JobWorkState` و`JobWorkFlow` و`JobRunner` عقدًا اختياريًا للـ jobs داخل الذاكرة. تحتفظ conversational Flows الحالية بسلوكها. لكل job هوية مستقلة ودور منشئ ومراجعة ومحاولة تنفيذ وحالة ومرحلة حالية ومراحل مكتملة وخطأ. تضيف التطبيقات مدخلات ومخرجات ذات أنواع محددة؛ ولا يلزم إنشاء فئة منفصلة للـ artifacts.
+
+عرّف حقول النتائج القابلة للكتابة في `output_fields`. لا يستطيع worker تغيير حقول المدخلات أو الملكية أو دورة الحياة عبر payload المخرجات. يُتحقق من السجل المرشح بالكامل قبل commit؛ ولا تغير التحديثات غير الصالحة أو التابعة لمالك آخر أو القديمة أو المكررة أو اللاحقة للانتهاء الحالة المقبولة.
+
+```python
+import asyncio
+from typing import ClassVar
+from pydantic import Field
+from crewai.flow import ConversationState, start
+from crewai.experimental.flow_jobs import (
+    JobRecord, JobRunner, JobState, JobWorkFlow, JobWorkState, add_job,
+)
+
+class ReportJob(JobRecord):
+    output_fields: ClassVar[frozenset[str]] = frozenset({"notes"})
+    question: str
+    notes: list[str] = Field(default_factory=list)
+    stage: str = "collect"
+
+class WorkChatState(ConversationState, JobState[ReportJob]):
+    pass
+
+class ReportWork(JobWorkFlow[JobWorkState[ReportJob]]):
+    @start()
+    def collect(self):
+        self.check_open()
+        self.publish_update("started")
+        self.publish_update(
+            "stage_completed", stage="collect", outputs={"notes": ["Example fact"]},
+        )
+        self.publish_update("completed")
+
+async def run_job():
+    state = WorkChatState()
+    changes = asyncio.Queue()
+
+    def make_work(job, inputs, publish):
+        return ReportWork(
+            initial_state=JobWorkState[ReportJob](job=job),
+            publish=publish, suppress_flow_events=True,
+        )
+
+    runner = JobRunner(state, make_work, on_update=changes.put)
+    try:
+        async with runner.state_lock:
+            job = ReportJob(session_id=state.id, question="Example report")
+            if not add_job(state, job):
+                raise RuntimeError("Job registration rejected")
+            runner.submit(job, {})
+        while True:
+            snapshot = await changes.get()
+            if snapshot["jobs"][0]["status"] in {"completed", "failed"}:
+                return snapshot
+    finally:
+        await runner.aclose()
+
+# Run in a script: asyncio.run(run_job())
+
+```
+
+أنشئ `JobRunner(state, make_work, on_update=publish_snapshot)` داخل asyncio loop يعمل بالفعل. تعيد `make_work(job, inputs, publish)` كائن `JobWorkFlow` غير متدفق، بحالة ذات أنواع محددة ونسخ من job والمدخلات، مع `publish=publish`. الدالة `on_update` هي async callback تتلقى snapshots متوافقة مع JSON (`session_id`, `seq`, `jobs`)؛ اربطها بوسيلة النقل الموجودة. يمكن لـ `format_error` حذف تفاصيل provider الحساسة، وتراقب `on_worker_event` بدء worker وانتهاءه.
+
+استخدم `runner.state_lock` مع **كل** دور في المقدمة وتسجيل job وتعديل لحالة الأب. سجّل بواسطة `add_job(state, job)` ثم استدعِ `runner.submit(job, inputs)`. شغّل الأدوار المتزامنة في worker thread مع الاحتفاظ بالقفل؛ ولا تشغّل دورين بالتوازي على Flow واحد. تعمل `publish_update()` أيضًا خارج event loop الخاص بالـ runner وتنتظر قبول commit قبل المرحلة التالية. يتطلب تحديث `stage_started` تثبيت المرحلة السابقة؛ ويحفظ `stage_completed` المخرجات ذات الأنواع المحددة قبل الإكمال الناجح. يحافظ الفشل اللاحق على المخرجات السابقة.
+
+قدّم حالة job ومخرجاته ذات الصلة عبر `build_router_context()` و`build_agent_context()`. لا تتحول snapshots تلقائيًا إلى رسائل محادثة عامة أو ردود صوتية. يبقى تفسير المجال وحدود قبول العمل من سياسات التطبيق.
+
+عند إغلاق الجلسة، استدعِ `runner.request_close()` على loop المالك قبل انتظار انتهاء التنفيذ في المقدمة، ثم `await runner.aclose()`. يؤدي ذلك إلى إغلاق قبول العمل والنشر وإرسال إشارة إلى كل worker وتحرير publishers المنتظرين وانتظار tasks المملوكة. يجب أن تستدعي طرق العمل `check_open()` عند حدود المراحل. لا تُجهض عملية provider قيد التنفيذ. يغلق فشل subscriber الـ runner بدل ترك publishers معلقين. لا ينبغي أن يؤدي إيقاف الكلام وحده إلى إغلاق runner.
+
+دورة الحياة الأولية هي `queued → running → completed/failed`. لا تتوفر pause/resume أو الاستبدال أو تسليم الردود تلقائيًا أو الاستعادة الموزعة. يمكن تسلسل السجلات باستخدام إمكانات الحالة الموجودة، لكن استعادتها لا تعيد إنشاء workers نشطين ولا تضمن تنفيذ الآثار الخارجية مرة واحدة بالضبط.

docs/edge/en/guides/flows/conversational-flows.mdx

@@ -617,3 +617,78 @@ from crewai.flow import (
 
 - [Mastering Flow State Management](/en/guides/flows/mastering-flow-state) — persistence, Pydantic state, `@persist`
 - [Build Your First Flow](/en/guides/flows/first-flow) — flow basics
+
+
+## Experimental jobs spanning conversational turns
+
+This API is experimental and may change. Import it explicitly from `crewai.experimental.flow_jobs`; it is not part of the stable `crewai.flow` surface.
+
+`JobRecord`, `JobState`, `JobUpdate`, `JobWorkState`, `JobWorkFlow`, and `JobRunner` provide an opt-in, in-memory job contract. Existing conversational Flows keep their behavior. A job has its own identity, originating turn, revision, execution attempt, status, current stage, completed stages, and error. Applications add typed inputs and outputs; no separate artifact class is required.
+
+Declare writable result fields in `output_fields`. Worker updates cannot change input, ownership, or lifecycle fields through an output payload. The complete candidate record is validated before a commit; malformed, foreign, stale, duplicate, and post-terminal updates leave accepted state unchanged.
+
+```python
+import asyncio
+from typing import ClassVar
+from pydantic import Field
+from crewai.flow import ConversationState, start
+from crewai.experimental.flow_jobs import (
+    JobRecord, JobRunner, JobState, JobWorkFlow, JobWorkState, add_job,
+)
+
+class ReportJob(JobRecord):
+    output_fields: ClassVar[frozenset[str]] = frozenset({"notes"})
+    question: str
+    notes: list[str] = Field(default_factory=list)
+    stage: str = "collect"
+
+class WorkChatState(ConversationState, JobState[ReportJob]):
+    pass
+
+class ReportWork(JobWorkFlow[JobWorkState[ReportJob]]):
+    @start()
+    def collect(self):
+        self.check_open()
+        self.publish_update("started")
+        self.publish_update(
+            "stage_completed", stage="collect", outputs={"notes": ["Example fact"]},
+        )
+        self.publish_update("completed")
+
+async def run_job():
+    state = WorkChatState()
+    changes = asyncio.Queue()
+
+    def make_work(job, inputs, publish):
+        return ReportWork(
+            initial_state=JobWorkState[ReportJob](job=job),
+            publish=publish, suppress_flow_events=True,
+        )
+
+    runner = JobRunner(state, make_work, on_update=changes.put)
+    try:
+        async with runner.state_lock:
+            job = ReportJob(session_id=state.id, question="Example report")
+            if not add_job(state, job):
+                raise RuntimeError("Job registration rejected")
+            runner.submit(job, {})
+        while True:
+            snapshot = await changes.get()
+            if snapshot["jobs"][0]["status"] in {"completed", "failed"}:
+                return snapshot
+    finally:
+        await runner.aclose()
+
+# Run in a script: asyncio.run(run_job())
+
+```
+
+Construct `JobRunner(state, make_work, on_update=publish_snapshot)` on a running asyncio loop. `make_work(job, inputs, publish)` returns a non-streaming `JobWorkFlow` with its own typed state, copied job/inputs, and `publish=publish`. `on_update` is an async callback receiving JSON-compatible snapshots (`session_id`, `seq`, `jobs`); adapt them to your existing transport. `format_error` can redact provider details, and `on_worker_event` observes worker start/settlement.
+
+Hold `runner.state_lock` for **every** foreground turn, job admission, and parent-state mutation. Register with `add_job(state, job)` and then call `runner.submit(job, inputs)`. Run synchronous turns in a worker thread while holding that lock; never run two turns concurrently on one Flow. Worker `publish_update()` also runs off the runner's event loop and waits for an accepted commit before the next stage. A `stage_started` update requires the previous stage to be committed; `stage_completed` retains typed outputs before successful completion. A later failure preserves earlier outputs.
+
+Project relevant job status and outputs through `build_router_context()` and `build_agent_context()`. Snapshots are not automatically public chat messages or spoken replies. Domain interpretation and admission capacity remain application policy.
+
+At session teardown, call `runner.request_close()` on its owning loop before draining foreground execution, then `await runner.aclose()`. This closes admission/publication, signals each worker, releases waiting publishers, and settles owned tasks. Work methods must call `check_open()` at stage boundaries. An in-flight provider operation is not aborted. Subscriber failure closes the runner rather than stranding publishers. Stopping speech alone should not close the runner.
+
+The initial lifecycle is `queued → running → completed/failed`. Pause/resume, replacement, automatic reply delivery, and distributed recovery are not supplied. Records can be serialized with existing state facilities, but restoring them does not recreate live workers or guarantee exactly-once external effects.

docs/edge/ko/guides/flows/conversational-flows.mdx

@@ -612,3 +612,78 @@ from crewai.flow import (
 
 - [Flow 상태 관리 마스터하기](/ko/guides/flows/mastering-flow-state)
 - [첫 Flow 만들기](/ko/guides/flows/first-flow)
+
+
+## 실험적 Jobs: 대화 턴에 걸친 실행
+
+이 API는 실험적이며 변경될 수 있습니다. `crewai.experimental.flow_jobs`에서 명시적으로 import합니다. 안정적인 `crewai.flow` API에 포함되지 않습니다.
+
+`JobRecord`, `JobState`, `JobUpdate`, `JobWorkState`, `JobWorkFlow`, `JobRunner`는 선택적으로 사용하는 메모리 기반 job 계약을 제공합니다. 기존 conversational Flow의 동작은 유지됩니다. Job은 독립적인 식별자, 시작 턴, revision, 실행 attempt, 상태, 현재 단계, 완료된 단계와 오류를 가집니다. 애플리케이션은 타입이 지정된 입력과 출력을 추가하며, 별도의 artifact 클래스는 필요하지 않습니다.
+
+쓰기 가능한 결과 필드는 `output_fields`에 선언합니다. Worker는 출력 payload를 통해 입력, 소유권 또는 lifecycle 필드를 변경할 수 없습니다. Commit 전에 전체 후보 record를 검증하며, 잘못된 형식, 다른 소유자, 오래된 revision/attempt, 중복 또는 종료 후 update는 수락된 상태를 바꾸지 않습니다.
+
+```python
+import asyncio
+from typing import ClassVar
+from pydantic import Field
+from crewai.flow import ConversationState, start
+from crewai.experimental.flow_jobs import (
+    JobRecord, JobRunner, JobState, JobWorkFlow, JobWorkState, add_job,
+)
+
+class ReportJob(JobRecord):
+    output_fields: ClassVar[frozenset[str]] = frozenset({"notes"})
+    question: str
+    notes: list[str] = Field(default_factory=list)
+    stage: str = "collect"
+
+class WorkChatState(ConversationState, JobState[ReportJob]):
+    pass
+
+class ReportWork(JobWorkFlow[JobWorkState[ReportJob]]):
+    @start()
+    def collect(self):
+        self.check_open()
+        self.publish_update("started")
+        self.publish_update(
+            "stage_completed", stage="collect", outputs={"notes": ["Example fact"]},
+        )
+        self.publish_update("completed")
+
+async def run_job():
+    state = WorkChatState()
+    changes = asyncio.Queue()
+
+    def make_work(job, inputs, publish):
+        return ReportWork(
+            initial_state=JobWorkState[ReportJob](job=job),
+            publish=publish, suppress_flow_events=True,
+        )
+
+    runner = JobRunner(state, make_work, on_update=changes.put)
+    try:
+        async with runner.state_lock:
+            job = ReportJob(session_id=state.id, question="Example report")
+            if not add_job(state, job):
+                raise RuntimeError("Job registration rejected")
+            runner.submit(job, {})
+        while True:
+            snapshot = await changes.get()
+            if snapshot["jobs"][0]["status"] in {"completed", "failed"}:
+                return snapshot
+    finally:
+        await runner.aclose()
+
+# Run in a script: asyncio.run(run_job())
+
+```
+
+실행 중인 asyncio loop에서 `JobRunner(state, make_work, on_update=publish_snapshot)`를 생성합니다. `make_work(job, inputs, publish)`는 독립적인 typed state, 복사된 job/입력, `publish=publish`를 사용하는 non-streaming `JobWorkFlow`를 반환합니다. `on_update`는 JSON 호환 snapshot (`session_id`, `seq`, `jobs`)을 받는 async callback이며 기존 transport에 연결합니다. `format_error`는 provider의 민감한 세부 정보를 제거할 수 있고, `on_worker_event`는 worker 시작과 종료를 관찰합니다.
+
+**모든** foreground turn, job 등록 및 부모 상태 변경에는 `runner.state_lock`을 사용합니다. `add_job(state, job)`으로 등록한 뒤 `runner.submit(job, inputs)`를 호출합니다. Lock을 유지한 상태에서 동기 turn을 worker thread로 실행하며 하나의 Flow에서 두 turn을 동시에 실행하지 않습니다. Worker의 `publish_update()`도 runner의 event loop 밖에서 실행되며 다음 단계 전에 commit 수락을 기다립니다. `stage_started` update는 이전 단계의 commit을 요구하며, `stage_completed`는 성공적으로 종료하기 전에 typed 출력을 보존합니다. 이후 실패해도 이전 출력은 유지됩니다.
+
+`build_router_context()`와 `build_agent_context()`를 통해 관련 job 상태와 출력을 제공합니다. Snapshot은 자동으로 공개 chat 메시지나 음성 응답이 되지 않습니다. Domain 해석과 작업 수용 한도는 애플리케이션 정책입니다.
+
+Session 종료 시 foreground 실행을 정리하기 전에 소유 loop에서 `runner.request_close()`를 호출하고 이후 `await runner.aclose()`를 실행합니다. 새 작업 등록과 publication을 닫고 각 worker에 신호를 보내며 대기 중인 publisher를 해제하고 소유 task를 정리합니다. Work method는 단계 경계에서 `check_open()`을 호출해야 합니다. 진행 중인 provider 작업은 강제로 중단되지 않습니다. Subscriber 실패 시 publisher가 무한히 기다리지 않도록 runner가 닫힙니다. 음성 재생만 중지할 때는 runner를 닫지 않습니다.
+
+초기 lifecycle은 `queued → running → completed/failed`입니다. Pause/resume, replacement, 자동 응답 전달과 분산 recovery는 제공하지 않습니다. 기존 state 기능으로 record를 직렬화할 수 있지만, 복원은 live worker를 다시 만들거나 외부 작업의 exactly-once 실행을 보장하지 않습니다.

docs/edge/pt-BR/guides/flows/conversational-flows.mdx

@@ -617,3 +617,78 @@ from crewai.flow import (
 
 - [Dominando o Gerenciamento de Estado em Flows](/pt-BR/guides/flows/mastering-flow-state) — persistência, estado Pydantic, `@persist`
 - [Construa Seu Primeiro Flow](/pt-BR/guides/flows/first-flow) — fundamentos de flow
+
+
+## Jobs experimentais que abrangem vários turnos de conversa
+
+Esta API é experimental e pode mudar. Importe-a explicitamente de `crewai.experimental.flow_jobs`; ela não faz parte da API estável de `crewai.flow`.
+
+`JobRecord`, `JobState`, `JobUpdate`, `JobWorkState`, `JobWorkFlow` e `JobRunner` oferecem um contrato opcional de jobs em memória. Os Flows conversacionais existentes mantêm seu comportamento. Um job possui identidade própria, turno de origem, revisão, tentativa de execução, status, etapa atual, etapas concluídas e erro. As aplicações adicionam entradas e saídas tipadas; não é necessária uma classe separada de artefatos.
+
+Declare os campos de resultado graváveis em `output_fields`. Updates do worker não podem alterar campos de entrada, propriedade ou ciclo de vida por meio do payload de saída. O registro candidato completo é validado antes do commit; updates inválidos, de outro proprietário, obsoletos, duplicados ou posteriores ao término não alteram o estado aceito.
+
+```python
+import asyncio
+from typing import ClassVar
+from pydantic import Field
+from crewai.flow import ConversationState, start
+from crewai.experimental.flow_jobs import (
+    JobRecord, JobRunner, JobState, JobWorkFlow, JobWorkState, add_job,
+)
+
+class ReportJob(JobRecord):
+    output_fields: ClassVar[frozenset[str]] = frozenset({"notes"})
+    question: str
+    notes: list[str] = Field(default_factory=list)
+    stage: str = "collect"
+
+class WorkChatState(ConversationState, JobState[ReportJob]):
+    pass
+
+class ReportWork(JobWorkFlow[JobWorkState[ReportJob]]):
+    @start()
+    def collect(self):
+        self.check_open()
+        self.publish_update("started")
+        self.publish_update(
+            "stage_completed", stage="collect", outputs={"notes": ["Example fact"]},
+        )
+        self.publish_update("completed")
+
+async def run_job():
+    state = WorkChatState()
+    changes = asyncio.Queue()
+
+    def make_work(job, inputs, publish):
+        return ReportWork(
+            initial_state=JobWorkState[ReportJob](job=job),
+            publish=publish, suppress_flow_events=True,
+        )
+
+    runner = JobRunner(state, make_work, on_update=changes.put)
+    try:
+        async with runner.state_lock:
+            job = ReportJob(session_id=state.id, question="Example report")
+            if not add_job(state, job):
+                raise RuntimeError("Job registration rejected")
+            runner.submit(job, {})
+        while True:
+            snapshot = await changes.get()
+            if snapshot["jobs"][0]["status"] in {"completed", "failed"}:
+                return snapshot
+    finally:
+        await runner.aclose()
+
+# Run in a script: asyncio.run(run_job())
+
+```
+
+Crie `JobRunner(state, make_work, on_update=publish_snapshot)` em um loop asyncio em execução. `make_work(job, inputs, publish)` retorna um `JobWorkFlow` sem streaming, com estado tipado próprio, cópias do job e das entradas, e `publish=publish`. `on_update` é um callback assíncrono que recebe snapshots compatíveis com JSON (`session_id`, `seq`, `jobs`); adapte-os ao transporte existente. `format_error` pode remover detalhes sensíveis do provider, e `on_worker_event` observa o início e o encerramento do worker.
+
+Use `runner.state_lock` para **todos** os turnos em primeiro plano, registros de jobs e alterações do estado pai. Registre com `add_job(state, job)` e então chame `runner.submit(job, inputs)`. Execute turnos síncronos em uma thread de trabalho mantendo esse lock; nunca execute dois turnos simultaneamente no mesmo Flow. O `publish_update()` do worker também é executado fora do event loop do runner e aguarda a aceitação do commit antes da próxima etapa. Um update `stage_started` exige que a etapa anterior tenha sido consolidada; `stage_completed` retém as saídas tipadas antes da conclusão bem-sucedida. Uma falha posterior preserva as saídas anteriores.
+
+Forneça status e saídas relevantes pelos hooks `build_router_context()` e `build_agent_context()`. Snapshots não se tornam automaticamente mensagens públicas ou respostas faladas. A interpretação do domínio e a capacidade de admissão continuam sendo políticas da aplicação.
+
+Ao encerrar a sessão, chame `runner.request_close()` no loop proprietário antes de aguardar a execução em primeiro plano e, depois, `await runner.aclose()`. Isso fecha a admissão e a publicação, sinaliza cada worker, libera publishers em espera e aguarda as tasks próprias. Os métodos de trabalho devem chamar `check_open()` nos limites das etapas. Uma operação do provider em andamento não é abortada. Uma falha do subscriber fecha o runner para não deixar publishers bloqueados. Interromper apenas a fala não deve fechar o runner.
+
+O ciclo inicial é `queued → running → completed/failed`. Pause/resume, substituição, entrega automática de respostas e recuperação distribuída não são fornecidos. Os registros podem ser serializados com os recursos de estado existentes, mas restaurá-los não recria workers ativos nem garante efeitos externos exatamente uma vez.

lib/crewai/src/crewai/experimental/flow_jobs.py

@@ -0,0 +1,415 @@
+"""Experimental job state and execution for work spanning Flow turns.
+
+This opt-in API is under active development; its names and contract may change.
+
+Job records are snapshots, not live executions. Applications extend JobRecord
+with typed input/output fields and serialize turns and commits with the runner's
+state_lock. No job output is automatically appended to public conversation history.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable, Callable
+from contextlib import suppress
+from copy import deepcopy
+from datetime import datetime, timezone
+from threading import Event, Lock
+from typing import Any, ClassVar, Generic, Literal, TypeVar
+from uuid import uuid4
+
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    JsonValue,
+    PrivateAttr,
+    ValidationError,
+)
+
+from crewai.flow.flow import Flow
+
+
+JobStatus = Literal["queued", "running", "completed", "failed"]
+JobUpdateKind = Literal[
+    "started", "stage_started", "stage_completed", "completed", "failed"
+]
+
+
+def _timestamp() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+class JobRecord(BaseModel):
+    """Serializable lifecycle identity; subclass to declare domain inputs/outputs."""
+
+    model_config = ConfigDict(extra="forbid")
+    output_fields: ClassVar[frozenset[str]] = frozenset()
+
+    job_id: str = Field(default_factory=lambda: str(uuid4()), min_length=1)
+    session_id: str = Field(min_length=1)
+    origin_turn_id: str = ""
+    kind: str = ""
+    revision: int = Field(default=1, gt=0, strict=True)
+    attempt: int = Field(default=1, gt=0, strict=True)
+    status: JobStatus = "queued"
+    stage: str = ""
+    committed_stages: list[str] = Field(default_factory=list)
+    last_update_seq: int = 0
+    error: str | None = None
+    created_at: str = Field(default_factory=_timestamp)
+    updated_at: str = Field(default_factory=_timestamp)
+
+
+JobT = TypeVar("JobT", bound=JobRecord)
+
+
+class JobState(BaseModel, Generic[JobT]):
+    """Opt-in state; combine with ConversationState in a custom state subclass."""
+
+    id: str = Field(default_factory=lambda: str(uuid4()))
+    jobs: dict[str, JobT] = Field(default_factory=dict)
+    job_sequence: int = 0
+
+
+class JobUpdate(BaseModel):
+    """Worker proposal, fenced by owner, revision, attempt and sequence.
+
+    Only stage_completed may set outputs. Output keys must name fields declared
+    as output_fields by the application's JobRecord subclass, never input or
+    lifecycle/identity fields.
+    The record's Pydantic schema validates the complete candidate before commit.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    session_id: str = Field(min_length=1)
+    job_id: str = Field(min_length=1)
+    revision: int = Field(gt=0, strict=True)
+    attempt: int = Field(gt=0, strict=True)
+    seq: int = Field(gt=0, strict=True)
+    kind: JobUpdateKind
+    stage: str = ""
+    outputs: dict[str, JsonValue] = Field(default_factory=dict)
+    error: str | None = None
+
+
+def add_job(state: JobState[Any], job: JobRecord) -> bool:
+    """Register one validated queued record through the serialized state path."""
+    if job.session_id != state.id or job.job_id in state.jobs or job.status != "queued":
+        return False
+    try:
+        candidate = type(job).model_validate(job.model_dump())
+        candidate.model_dump_json()  # Reject live/non-serializable application fields.
+    except (ValidationError, ValueError, TypeError):
+        return False
+    state.jobs[job.job_id] = job
+    state.job_sequence += 1
+    return True
+
+
+def commit_job_update(state: JobState[Any], update: JobUpdate) -> bool:
+    """Atomically accept a valid update; invalid proposals leave state unchanged.
+
+    This function does not acquire a lock. Its caller must share one serialized
+    mutation path with foreground turns and job admission (JobRunner.state_lock).
+    A stage must commit before another stage or successful completion is admitted.
+    """
+    job = state.jobs.get(update.job_id)
+    if (
+        job is None
+        or update.session_id != state.id
+        or job.session_id != update.session_id
+        or (job.revision, job.attempt) != (update.revision, update.attempt)
+        or update.seq <= job.last_update_seq
+        or job.status in {"completed", "failed"}
+    ):
+        return False
+    if update.outputs and update.kind != "stage_completed":
+        return False
+    changes: dict[str, Any] = {}
+    if update.kind == "started" and job.status == "queued":
+        changes["status"] = "running"
+    elif update.kind == "stage_started" and job.status == "running":
+        if (
+            not update.stage
+            or update.stage in job.committed_stages
+            or (job.stage and job.stage not in job.committed_stages)
+        ):
+            return False
+        changes["stage"] = update.stage
+    elif update.kind == "stage_completed" and job.status == "running":
+        if (
+            not update.stage
+            or update.stage != job.stage
+            or update.stage in job.committed_stages
+        ):
+            return False
+        output_fields = job.output_fields & (
+            set(type(job).model_fields) - set(JobRecord.model_fields)
+        )
+        if not update.outputs.keys() <= output_fields:
+            return False
+        changes.update(update.outputs)
+        changes["committed_stages"] = [*job.committed_stages, update.stage]
+    elif update.kind == "completed" and job.status == "running":
+        if job.stage and job.stage not in job.committed_stages:
+            return False
+        changes["status"] = "completed"
+    elif update.kind == "failed":
+        changes.update(status="failed", error=update.error or "Job failed.")
+    else:
+        return False
+    changes.update(last_update_seq=update.seq, updated_at=_timestamp())
+    try:
+        candidate = type(job).model_validate({**job.model_dump(), **changes})
+        candidate.model_dump_json()
+    except (ValidationError, ValueError, TypeError):
+        return False
+    # Preserve record identity for existing consumers, after all validation.
+    for name in type(job).model_fields:
+        setattr(job, name, getattr(candidate, name))
+    state.job_sequence += 1
+    return True
+
+
+class JobWorkState(BaseModel, Generic[JobT]):
+    """Worker-owned snapshot; subclass with stage inputs and local results."""
+
+    job: JobT
+    seq: int = 0
+
+
+WorkStateT = TypeVar("WorkStateT", bound=JobWorkState[Any])
+UpdatePublisher = Callable[[JobUpdate], bool]
+
+
+class JobWorkFlow(Flow[WorkStateT]):
+    """Separate execution with acknowledged publication and boundary-only stop."""
+
+    _publish: UpdatePublisher = PrivateAttr()
+    _stop: Event = PrivateAttr(default_factory=Event)
+
+    def __init__(self, *, publish: UpdatePublisher, **kwargs: Any) -> None:
+        super().__init__(**kwargs)
+        self._publish = publish
+
+    def request_stop(self) -> None:
+        """Prevent later stages after an in-flight operation settles."""
+        self._stop.set()
+
+    def check_open(self) -> None:
+        """Call at stage boundaries; this cannot abort an arbitrary provider call."""
+        if self._stop.is_set():
+            raise RuntimeError("Job stopped at session close.")
+
+    def publish_update(self, kind: JobUpdateKind, **details: Any) -> None:
+        """Wait for parent acceptance before executing a dependent stage."""
+        self.state.seq += 1
+        job = self.state.job
+        update = JobUpdate(
+            session_id=job.session_id,
+            job_id=job.job_id,
+            revision=job.revision,
+            attempt=job.attempt,
+            seq=self.state.seq,
+            kind=kind,
+            **details,
+        )
+        if not self._publish(update):
+            raise RuntimeError("Job session closed or update rejected.")
+
+
+class JobRunner:
+    """Session-owned in-process bridge; construct on a running asyncio loop.
+
+    make_work receives copied inputs and a synchronous publisher for worker
+    threads. on_update receives serializable snapshots outside state_lock.
+    Foreground turns must also hold state_lock. Neither callbacks nor workers
+    may mutate parent state directly. Snapshot restoration does not restart work.
+    """
+
+    def __init__(
+        self,
+        state: JobState[Any],
+        make_work: Callable[[JobRecord, Any, UpdatePublisher], JobWorkFlow[Any]],
+        *,
+        on_update: Callable[[dict[str, Any]], Awaitable[None]],
+        on_worker_event: Callable[[str, JobRecord], None] | None = None,
+        format_error: Callable[[Exception], str] = str,
+    ) -> None:
+        self.state = state
+        self.make_work = make_work
+        self.on_update = on_update
+        self.on_worker_event = on_worker_event
+        self.format_error = format_error
+        self.state_lock = asyncio.Lock()
+        self.closed = False
+        self.tasks: dict[str, asyncio.Task[None]] = {}
+        self.workflows: dict[str, JobWorkFlow[Any]] = {}
+        self._loop = asyncio.get_running_loop()
+        self._events: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
+        self._receipt_lock = Lock()
+        self._receipts: set[Event] = set()
+        self._event_task = asyncio.create_task(self._process_events())
+        self._event_task.add_done_callback(lambda _: self.request_close())
+
+    def snapshot(self) -> dict[str, Any]:
+        """Read under state_lock when work can still mutate state."""
+        return {
+            "session_id": self.state.id,
+            "seq": self.state.job_sequence,
+            "jobs": [job.model_dump(mode="json") for job in self.state.jobs.values()],
+        }
+
+    def submit(self, job: JobRecord, inputs: Any) -> None:
+        """Schedule an already-registered job; safe from a foreground thread."""
+        with self._receipt_lock:
+            if self.closed:
+                raise RuntimeError("Job runner is closed.")
+            self._loop.call_soon_threadsafe(
+                self._events.put_nowait,
+                {
+                    "kind": "admit",
+                    "job": job.model_copy(deep=True),
+                    "inputs": deepcopy(inputs),
+                },
+            )
+
+    def _accept_update(self, update: JobUpdate) -> bool:
+        if self._loop is _running_loop():
+            raise RuntimeError(
+                "Publish job updates from a worker thread, not the runner's event loop."
+            )
+        settled = Event()
+        receipt = {"accepted": False}
+        with self._receipt_lock:
+            if self.closed:
+                return False
+            self._receipts.add(settled)
+            self._loop.call_soon_threadsafe(
+                self._events.put_nowait,
+                {
+                    "kind": "update",
+                    "update": update,
+                    "settled": settled,
+                    "receipt": receipt,
+                },
+            )
+        try:
+            settled.wait()
+            return receipt["accepted"]
+        finally:
+            with self._receipt_lock:
+                self._receipts.discard(settled)
+
+    async def _process_events(self) -> None:
+        while True:
+            event = await self._events.get()
+            if event["kind"] == "admit":
+                job = event["job"]
+                async with self.state_lock:
+                    current = self.state.jobs.get(job.job_id)
+                    if (
+                        self.closed
+                        or current is None
+                        or current.status != "queued"
+                        or current.session_id != job.session_id
+                        or (current.revision, current.attempt)
+                        != (job.revision, job.attempt)
+                        or job.job_id in self.tasks
+                    ):
+                        continue
+                    admission_snapshot = self.snapshot()
+                await self.on_update(admission_snapshot)
+                if self.closed:
+                    continue
+                try:
+                    work = self.make_work(
+                        job.model_copy(deep=True), event["inputs"], self._accept_update
+                    )
+                    if not isinstance(work, JobWorkFlow) or work.stream:
+                        raise ValueError(
+                            "JobRunner requires a non-streaming JobWorkFlow; publish stage updates instead."
+                        )
+                    work_job = work.state.job
+                    if (
+                        work_job.session_id,
+                        work_job.job_id,
+                        work_job.revision,
+                        work_job.attempt,
+                    ) != (job.session_id, job.job_id, job.revision, job.attempt):
+                        raise ValueError(
+                            "Worker identity does not match the admitted job."
+                        )
+                    self.workflows[job.job_id] = work
+                    self.tasks[job.job_id] = asyncio.create_task(
+                        asyncio.to_thread(self._run, work)
+                    )
+                except Exception as exc:
+                    async with self.state_lock:
+                        accepted = commit_job_update(
+                            self.state,
+                            JobUpdate(
+                                session_id=job.session_id,
+                                job_id=job.job_id,
+                                revision=job.revision,
+                                attempt=job.attempt,
+                                seq=1,
+                                kind="failed",
+                                error=self.format_error(exc),
+                            ),
+                        )
+                        snapshot = self.snapshot() if accepted else None
+                    if snapshot is not None:
+                        await self.on_update(snapshot)
+            else:
+                try:
+                    async with self.state_lock:
+                        accepted = not self.closed and commit_job_update(
+                            self.state, event["update"]
+                        )
+                        snapshot = self.snapshot() if accepted else None
+                    # Acceptance is independent of whether a transport succeeds.
+                    event["receipt"]["accepted"] = accepted
+                    if snapshot is not None:
+                        await self.on_update(snapshot)
+                finally:
+                    event["settled"].set()
+
+    def _run(self, work: JobWorkFlow[Any]) -> None:
+        job = work.state.job
+        try:
+            if self.on_worker_event:
+                self.on_worker_event("started", job)
+            work.kickoff()
+        except Exception as exc:
+            with suppress(RuntimeError):
+                work.publish_update("failed", error=self.format_error(exc))
+        finally:
+            if self.on_worker_event:
+                self.on_worker_event("settled", job)
+
+    def request_close(self) -> None:
+        """Close admission/publication before draining foreground execution."""
+        with self._receipt_lock:
+            self.closed = True
+            for receipt in self._receipts:
+                receipt.set()
+        for work in self.workflows.values():
+            work.request_stop()
+
+    async def aclose(self) -> None:
+        """Settle owned workers; in-flight provider calls may still take time."""
+        self.request_close()
+        await asyncio.gather(*self.tasks.values(), return_exceptions=True)
+        self._event_task.cancel()
+        with suppress(asyncio.CancelledError):
+            await self._event_task
+
+
+def _running_loop() -> asyncio.AbstractEventLoop | None:
+    try:
+        return asyncio.get_running_loop()
+    except RuntimeError:
+        return None

lib/crewai/tests/test_flow_jobs.py

@@ -0,0 +1,325 @@
+"""Job contracts exercised without a transport or external providers."""
+
+from __future__ import annotations
+
+import asyncio
+from threading import Event
+from typing import ClassVar
+
+from pydantic import Field, PrivateAttr, ValidationError
+import pytest
+
+from crewai.flow import ConversationState, Flow, listen, start
+from crewai.experimental.flow_jobs import (
+    JobRecord, JobRunner, JobState, JobUpdate, JobWorkFlow, JobWorkState,
+    add_job, commit_job_update,
+)
+
+
+class ReportJob(JobRecord):
+    output_fields: ClassVar[frozenset[str]] = frozenset({"notes", "answer"})
+    question: str
+    stage: str = "collect"
+    notes: list[str] = Field(default_factory=list)
+    answer: str = ""
+
+
+class ReportState(ConversationState, JobState[ReportJob]):
+    pass
+
+
+class ReportWorkState(JobWorkState[ReportJob]):
+    pass
+
+
+class ReportWork(JobWorkFlow[ReportWorkState]):
+    _entered: Event = PrivateAttr()
+    _release: Event = PrivateAttr()
+    _fail: bool = PrivateAttr()
+
+    def __init__(self, job, publish, entered, release, fail=True):
+        super().__init__(
+            publish=publish,
+            initial_state=ReportWorkState(job=job),
+            suppress_flow_events=True,
+        )
+        self._entered, self._release, self._fail = entered, release, fail
+
+    @start()
+    def collect(self):
+        self.check_open()
+        self.publish_update("started")
+        self._entered.set()
+        if not self._release.wait(5):
+            raise RuntimeError("Test gate not released")
+        self.publish_update(
+            "stage_completed", stage="collect", outputs={"notes": ["Retained fact"]}
+        )
+
+    @listen(collect)
+    def write(self):
+        self.check_open()
+        self.publish_update("stage_started", stage="write")
+        if self._fail:
+            raise RuntimeError("Writing failed")
+        self.publish_update(
+            "stage_completed", stage="write", outputs={"answer": "Finished report"}
+        )
+        self.publish_update("completed")
+
+
+class StatusFlow(Flow[ReportState]):
+    conversational = True
+
+    def route_turn(self, context):
+        return "converse"
+
+    def build_router_context(self):
+        return {
+            **super().build_router_context(),
+            "jobs": [j.model_dump() for j in self.state.jobs.values()],
+        }
+
+    @listen("converse")
+    def converse_turn(self):
+        job = next(iter(self.state.jobs.values()))
+        reply = f"{job.status}: {', '.join(job.notes)}"
+        self.append_assistant_message(reply)
+        return reply
+
+
+def proposal(state, job, seq, kind, **details):
+    return JobUpdate(
+        session_id=state.id,
+        job_id=job.job_id,
+        revision=job.revision,
+        attempt=job.attempt,
+        seq=seq,
+        kind=kind,
+        **details,
+    )
+
+
+def test_typed_commits_fence_identity_protect_inputs_and_retain_partial_outputs():
+    state = ReportState()
+    job = ReportJob(session_id=state.id, question="Write a report")
+    assert add_job(state, job)
+    assert not add_job(state, job)
+    assert not add_job(state, ReportJob(session_id="foreign", question="Other"))
+    assert commit_job_update(state, proposal(state, job, 1, "started"))
+    before = state.model_dump_json()
+    good = proposal(
+        state, job, 2, "stage_completed", stage="collect", outputs={"notes": ["Fact"]}
+    )
+    bad_updates = [
+        good.model_copy(update={"session_id": "foreign"}),
+        good.model_copy(update={"job_id": "missing"}),
+        good.model_copy(update={"revision": 2}),
+        good.model_copy(update={"attempt": 2}),
+        good.model_copy(update={"stage": "wrong"}),
+        good.model_copy(update={"outputs": {"notes": "Invalid list"}}),
+        good.model_copy(update={"outputs": {"question": "Rewrite immutable input"}}),
+        good.model_copy(update={"outputs": {"status": "completed"}}),
+        proposal(state, job, 2, "stage_started", stage="write"),
+        proposal(state, job, 2, "completed"),
+    ]
+    for update in bad_updates:
+        assert not commit_job_update(state, update)
+        assert state.model_dump_json() == before
+    assert commit_job_update(state, good)
+    assert not commit_job_update(state, good)
+    assert not commit_job_update(state, good.model_copy(update={"seq": 3}))
+    assert commit_job_update(
+        state, proposal(state, job, 3, "stage_started", stage="write")
+    )
+    assert commit_job_update(
+        state, proposal(state, job, 4, "failed", error="Writing failed")
+    )
+    assert not commit_job_update(state, proposal(state, job, 5, "completed"))
+    assert job.notes == ["Fact"]
+    assert job.question == "Write a report"
+    assert job.status == "failed"
+    assert job.committed_stages == ["collect"]
+    assert state.job_sequence == 5
+    restored = ReportState.model_validate_json(state.model_dump_json())
+    assert type(restored.jobs[job.job_id]) is ReportJob
+    assert restored == state
+
+
+def test_success_requires_the_current_stage_to_commit_and_updates_are_serializable():
+    state = ReportState()
+    job = ReportJob(session_id=state.id, question="Report")
+    assert add_job(state, job)
+    assert commit_job_update(state, proposal(state, job, 1, "started"))
+    assert commit_job_update(
+        state,
+        proposal(
+            state,
+            job,
+            2,
+            "stage_completed",
+            stage="collect",
+            outputs={"notes": ["Fact"]},
+        ),
+    )
+    assert commit_job_update(state, proposal(state, job, 3, "completed"))
+    assert not commit_job_update(
+        state, proposal(state, job, 4, "failed", error="Late failure")
+    )
+    assert job.status == "completed"
+    with pytest.raises(ValidationError):
+        proposal(
+            state,
+            job,
+            4,
+            "stage_completed",
+            stage="collect",
+            outputs={"notes": Event()},
+        )
+    with pytest.raises(ValidationError):
+        proposal(state, job, True, "started")
+
+
+@pytest.mark.asyncio
+async def test_runner_allows_turns_while_held_and_retains_outputs_after_failure():
+    flow = StatusFlow(suppress_flow_events=True)
+    entered, release = Event(), Event()
+    snapshots = asyncio.Queue()
+    made = []
+
+    def make(job, inputs, publish):
+        inputs.append("Worker copy")
+        work = ReportWork(job, publish, entered, release)
+        made.append(work)
+        return work
+
+    runner = JobRunner(flow.state, make, on_update=snapshots.put)
+    inputs = ["Original"]
+    job = ReportJob(session_id=flow.state.id, question="Report")
+    try:
+        async with runner.state_lock:
+            assert add_job(flow.state, job)
+            runner.submit(job, inputs)
+            runner.submit(
+                job, inputs
+            )  # Duplicate admission must not launch work twice.
+        assert await asyncio.to_thread(entered.wait, 5)
+        async with runner.state_lock:
+            reply = await asyncio.to_thread(flow.handle_turn, "Are you still working?")
+            assert reply == "running: "
+            assert flow.build_router_context()["jobs"][0]["status"] == "running"
+        assert inputs == ["Original"]
+        assert made[0].state.job is not job
+        release.set()
+        while True:
+            snapshot = await asyncio.wait_for(snapshots.get(), 5)
+            if snapshot["jobs"][0]["status"] == "failed":
+                break
+        async with runner.state_lock:
+            assert (
+                await asyncio.to_thread(flow.handle_turn, "What did you find?")
+                == "failed: Retained fact"
+            )
+        assert len(made) == 1
+        assert job.error == "Writing failed"
+        assert len(flow.state.messages) == 4
+    finally:
+        release.set()
+        await runner.aclose()
+        flow.finalize_session_traces(discard=True)
+    assert all(task.done() for task in runner.tasks.values())
+    assert not runner._receipts
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("failure", ["raise", "wrong_owner", "streaming"])
+async def test_worker_start_failure_is_a_visible_terminal_record(failure):
+    state = ReportState()
+    snapshots = asyncio.Queue()
+
+    entered, release = Event(), Event()
+
+    def fail(job, inputs, publish):
+        if failure == "raise":
+            raise RuntimeError("Worker unavailable")
+        work = ReportWork(job, publish, entered, release)
+        if failure == "wrong_owner":
+            work.state.job.session_id = "foreign"
+        else:
+            work.stream = True
+        return work
+
+    runner = JobRunner(state, fail, on_update=snapshots.put)
+    job = ReportJob(session_id=state.id, question="Report")
+    try:
+        async with runner.state_lock:
+            assert add_job(state, job)
+            runner.submit(job, [])
+        await asyncio.wait_for(snapshots.get(), 5)
+        failed = await asyncio.wait_for(snapshots.get(), 5)
+        assert failed["jobs"][0]["status"] == "failed"
+        assert job.error
+        assert not entered.is_set()
+        assert job.session_id == state.id
+    finally:
+        await runner.aclose()
+
+
+@pytest.mark.asyncio
+async def test_close_releases_pending_receipts_before_foreground_lock_is_released():
+    state = ReportState()
+    entered, release = Event(), Event()
+    runner = JobRunner(
+        state,
+        lambda job, inputs, publish: ReportWork(job, publish, entered, release),
+        on_update=asyncio.Queue().put,
+    )
+    job = ReportJob(session_id=state.id, question="Report")
+    async with runner.state_lock:
+        assert add_job(state, job)
+        runner.submit(job, [])
+    try:
+        assert await asyncio.to_thread(entered.wait, 5)
+        async with runner.state_lock:
+            release.set()
+            runner.request_close()
+            await asyncio.wait_for(runner.aclose(), 5)
+        assert job.notes == []
+        assert all(task.done() for task in runner.tasks.values())
+        assert not runner._receipts
+        with pytest.raises(RuntimeError, match="closed"):
+            runner.submit(job, [])
+    finally:
+        release.set()
+        await runner.aclose()
+
+
+@pytest.mark.asyncio
+async def test_failed_update_subscriber_does_not_leave_a_worker_waiting_forever():
+    state = ReportState()
+    entered, release = Event(), Event()
+    release.set()
+
+    async def broken(snapshot):
+        if snapshot["jobs"][0]["status"] == "running":
+            raise RuntimeError("Subscriber failed")
+
+    runner = JobRunner(
+        state,
+        lambda job, inputs, publish: ReportWork(job, publish, entered, release),
+        on_update=broken,
+    )
+    job = ReportJob(session_id=state.id, question="Report")
+    try:
+        async with runner.state_lock:
+            assert add_job(state, job)
+            runner.submit(job, [])
+        with pytest.raises(RuntimeError, match="Subscriber failed"):
+            await asyncio.wait_for(runner._event_task, 5)
+        with pytest.raises(RuntimeError, match="Subscriber failed"):
+            await asyncio.wait_for(runner.aclose(), 5)
+        assert runner.closed
+        assert all(task.done() for task in runner.tasks.values())
+        assert not runner._receipts
+    finally:
+        runner.request_close()