mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-10-12 02:12:31 +00:00
feat(flow): queue relevant background replies (OSS-192) (#7911)
* feat(flow): queue relevant background replies (OSS-192) * fix(flow): fence queued output after shared reply interruption * fix(flow): keep prepared delivery retries side effect free
This commit is contained in:
Lorenze Jay
@@ -745,3 +745,63 @@ runner.interrupt(
|
||||
لا يوفر هذا الإصدار queues للردود الخلفية أو جدولة دور الكلام أو إيصالات التشغيل أو محاذاة الكلمات الدقيقة أو speculative generation أو pause/resume للـ jobs. تحتوي snapshots على سجلات لا أقفال أو إشارات إلغاء أو تنفيذ نشط، ولا تستأنف العمل أو تعيد تشغيل الرد. استعد الحالة قبل بدء دور حي متتبع جديد؛ ويرفض wrapper خيار `from_checkpoint` و`restore_from_state_id` عمدًا. تبقى المصالحة المستمرة وحفظ سجلات التتبع النهائية عملًا منفصلًا. استخدم واجهات Flow المستقرة لمسارات checkpoint الموجودة.
|
||||
|
||||
أثناء التنفيذ المتتبع، تظل الحالة المستعادة الحالية هي المرجع: يُمنع التحميل التلقائي لحالة الجلسة كي لا تُفقد سجلات turn/reply المقبولة أو تحديثات jobs الخلفية. تحتفظ واجهات Flow الأصلية بسلوك تحميل الحالة المحفوظة المعتاد. لا يؤدي إغلاق iterator بعد حدث نهائي إلى مقاطعة الرد المكتمل. ويظل حدث الفشل إشارة نهائية مقبولة لدورة الحياة إذا وصلت المقاطعة بعد الفشل.
|
||||
|
||||
## الردود الخلفية التجريبية المنتظرة
|
||||
|
||||
تضيف `crewai.experimental.flow_replies` الأنواع `ReplyQueueState` و`ReplyQueue` و`DeliveryRecord` و`DeliveryFloor` و`CoveredUpdate` و`ClientActivity` فوق عقد الأدوار والردود. اجمع `ReplyQueueState` مع `JobState[MyJob]` وأنشئ `TurnRunner` **واحدًا** و`ReplyQueue` **واحدًا** لكل Flow حي. تبقى الواجهات اختيارية ولا يتغير منفذ jobs أو واجهات المحادثة المستقرة.
|
||||
|
||||
تتوفر نتائج job الناجح في الحالة فورًا، حتى إن انتظر الإعلان. تقبل `enqueue(job)` فقط job حاليًا ملتزمًا بحالة `completed` وله دور أصلي معروف. يسجل التسليم revision وattempt وتسلسل التحديث المقبول؛ تندمج الإشعارات المكررة في `delivery_id` نفسه، وهو أيضًا هوية `ReplyRecord` العام لاحقًا. يحصل كل إنجاز job على إعلان قابل للإسناد، ولا تنشئ تحديثات المراحل المتوسطة طابور كلام.
|
||||
|
||||
```python
|
||||
from crewai.experimental.flow_jobs import JobState
|
||||
from crewai.experimental.flow_turns import TurnRunner
|
||||
from crewai.experimental.flow_replies import (
|
||||
ClientActivity, CoveredUpdate, ReplyQueue, ReplyQueueState,
|
||||
)
|
||||
|
||||
class WorkChatState(ReplyQueueState, JobState[MyJob]):
|
||||
pass
|
||||
|
||||
# Your conversational Flow uses WorkChatState; jobs is its JobRunner.
|
||||
turns = TurnRunner(flow)
|
||||
replies = ReplyQueue(turns, is_relevant=lambda job: job.job_id not in cancelled_job_ids)
|
||||
|
||||
async def on_job_snapshot(snapshot):
|
||||
async with jobs.state_lock:
|
||||
for job in flow.state.jobs.values():
|
||||
if job.status == "completed":
|
||||
replies.enqueue(job)
|
||||
# Publish snapshots and wake your application-owned delivery task.
|
||||
|
||||
async def try_delivery():
|
||||
async with jobs.state_lock:
|
||||
delivery = replies.claim()
|
||||
if delivery is None:
|
||||
return
|
||||
events = replies.prepare(
|
||||
delivery.delivery_id,
|
||||
lambda current_jobs: "\n\n".join(job.answer for job in current_jobs),
|
||||
)
|
||||
for event in events:
|
||||
async with jobs.state_lock:
|
||||
if not replies.accepts_output(delivery.delivery_id):
|
||||
replies.settle(delivery.delivery_id, "skipped")
|
||||
return
|
||||
if turns.accept_event(event):
|
||||
await send_public_event(event)
|
||||
# Generation has completed. The speaking floor is still leased.
|
||||
# Validate whole-message adapter feedback before calling:
|
||||
# replies.settle(delivery.delivery_id, "completed")
|
||||
```
|
||||
|
||||
سلسل enqueue وclaim وإعداد النص وتغييرات التغطية وتسليم المخرجات باستخدام `JobRunner.state_lock` نفسه المستخدم للأدوار والتحديثات. يحمي `ReplyQueue` أوامره بقفل إضافي لمنع استحواذ claim متزامنين على دور الكلام. تقرأ `prepare()` نسخًا من **المخرجات الحالية المقبولة**، لا نصًا محفوظًا عند enqueue، وتصدر أحداث `started`/`text`/`completed` العامة مع هوية الأصل وjob. لا تنفذ LLM أو تضيف تاريخًا أو تجدول TTS؛ يختار التطبيق النص العام والتاريخ ومهام التسليم والنقل.
|
||||
|
||||
تحجز `set_foreground(True)` البوابة خلال التوليد والتصفية في المقدمة، بما في ذلك طلبات المستخدم المنتظرة؛ حررها بعد الاستقرار. تقبل `observe_client(ClientActivity(...))` الجلسة المالكة وتسلسل client المتزايد فقط، وقيمًا منطقية صارمة لـ `recording` و`playback` و`muted`. يجب أن يكون التشغيل المحدد تابعًا لـ Flow؛ ولا يمسح توقف قديم لرد مختلف تشغيلًا أحدث. تسمح `delivery_id` الفارغة بإقرار صوتي محلي محفوظ. يمنع التسجيل أو التشغيل أو الكتم أو الدور الجاري أو إغلاق الجلسة أو تسليم نشط `claim()`. إذا بدأ التسجيل/الكتم أو وصل طلب مستخدم أثناء الإعلان، فعلى adapter مقاطعة التسليم وإيقافه ثم تحديث النشاط وإيقاظ الجدولة.
|
||||
|
||||
تغير `claim()` تسليمًا مؤهلًا واحدًا ذريًا من `pending` إلى `scheduled`. قبل الإعداد و**كل** تسليم، تحقق عبر `accepts_output(delivery_id)` واقبل أحداث النص بواسطة `turns.accept_event(event)`. تغيّر الهوية/الإصدار أو غياب jobs أو عدم نجاحها أو انتهاء الجلسة أو تغطية التحديث أو سياسة `is_relevant(job)` يبطل الإعلان. يجب أن تكون callback بلا آثار جانبية؛ استخدمها لسياسة الإلغاء والاستبدال الخاصة بالتطبيق خارج عقد jobs الحالي. أيقظ مهمة التسليم عند تغير النشاط أو الصلة. لا تحذف التجاوزات المخرجات المحفوظة.
|
||||
|
||||
بعد تسليم ملخص مطلوب فعليًا، تسجل `mark_covered(CoveredUpdate.from_job(job))` الإصدار المقبول وتتجاوز الإعلانات المطابقة المنتظرة. مجرد طلب الملخص ليس تسليمًا. يظل إصدار job اللاحق مؤهلًا. لا تعتبر كل jobs في السياق مغطاة برد حالة أو محادثة غير مرتبطة.
|
||||
|
||||
تحرر `settle(delivery_id, status, reason=...)` التسليم النشط فقط ومرة واحدة. الحالات النهائية هي `completed` و`interrupted` و`skipped` و`failed`؛ اكتمال التوليد وحده **لا** ينهي التسليم. تحقق من الجلسة والدور وinput revision وهوية التسليم عند حدود adapter، واقبل الاكتمال بعد استقرار التوليد فقط. اكتمال الرسالة عند adapter ملاحظة جدولة لا إثبات للكلمات المسموعة. عند المقاطعة أو الفشل أو الإبطال، أوقف التشغيل المحلي والتسليم الجاري واحجب المخرجات المتبقية وأبق النتيجة متاحة. لا تعاد الإعلانات المقطوعة/الفاشلة تلقائيًا؛ يمكن طلب ملخص جديد. لا يغيّر فشل TTS نجاح job.
|
||||
|
||||
استدع `close()` قبل إلغاء وتصفية مهام التسليم التابعة للتطبيق. يحجب الردود المنتظرة/النشطة ويحفظ مخرجات jobs. تتضمن snapshot ملاحظات الدور وسجلات التسليم، لا مؤقتًا أو اتصالًا صوتيًا أو حجزًا قابلًا للاستئناف. لا تعيد الاستعادة تشغيل التسليم؛ سوّ السجلات النشطة وفق سياسة استرداد التطبيق. تبقى إيصالات المقاطع والسياق المنطوق الدقيق والإلغاء/الاستبدال التلقائي للـ jobs والجدولة الموزعة ومهل التشغيل ميزات منفصلة. العميل الذي لا يبلغ الاكتمال يحافظ تحفظيًا على الحجز حتى المقاطعة أو الإغلاق؛ يمكن للتطبيق إضافة سياسة مهلة صريحة.
|
||||
|
||||
@@ -746,3 +746,63 @@ To compose with background work, declare `class WorkChatState(TurnState, JobStat
|
||||
This increment does not supply background reply queues, speaking-floor scheduling, playback receipts, exact word alignment, speculative generation, or job pause/resume. Snapshots contain records, not locks, cancellation signals or running executions. They do not resume work or replay delivery. Restore state before creating a new tracked live turn; `from_checkpoint` and `restore_from_state_id` are deliberately rejected by this wrapper. Durable reconciliation and persistence of terminal tracking records remain separate work. Continue using the stable Flow APIs for existing checkpoint workflows.
|
||||
|
||||
During tracked execution, the current restored state remains authoritative: automatic session reload is suppressed so it cannot discard admitted turn/reply records or accepted background job updates. Native Flow turn APIs retain their existing persistence reload behavior. Closing after a terminal event does not interrupt the completed reply. Failure remains an acceptable terminal lifecycle signal if interruption arrives after failure.
|
||||
|
||||
## Experimental queued background replies
|
||||
|
||||
`crewai.experimental.flow_replies` adds `ReplyQueueState`, `ReplyQueue`, `DeliveryRecord`, `DeliveryFloor`, `CoveredUpdate`, and `ClientActivity` on top of the turn/reply contract. Declare state that combines `ReplyQueueState` and `JobState[MyJob]`, then create **one** `TurnRunner` and **one** `ReplyQueue` for that live Flow. Existing Flows remain opt-in; the job executor and stable conversational APIs do not change.
|
||||
|
||||
A successful job is immediately available in state, even when its announcement waits. `enqueue(job)` accepts only a current, committed `completed` job with an existing originating turn. The pending delivery stores that job's revision, attempt and accepted update sequence; duplicate notifications coalesce to the same `delivery_id`. This ID is also the eventual public `ReplyRecord` identity. Each job completion gets one attributable announcement; intermediate stage progress does not create a speaking queue.
|
||||
|
||||
```python
|
||||
from crewai.experimental.flow_jobs import JobState
|
||||
from crewai.experimental.flow_turns import TurnRunner
|
||||
from crewai.experimental.flow_replies import (
|
||||
ClientActivity, CoveredUpdate, ReplyQueue, ReplyQueueState,
|
||||
)
|
||||
|
||||
class WorkChatState(ReplyQueueState, JobState[MyJob]):
|
||||
pass
|
||||
|
||||
# Your conversational Flow uses WorkChatState; jobs is its JobRunner.
|
||||
turns = TurnRunner(flow)
|
||||
replies = ReplyQueue(turns, is_relevant=lambda job: job.job_id not in cancelled_job_ids)
|
||||
|
||||
async def on_job_snapshot(snapshot):
|
||||
async with jobs.state_lock:
|
||||
for job in flow.state.jobs.values():
|
||||
if job.status == "completed":
|
||||
replies.enqueue(job)
|
||||
# Publish snapshots and wake your application-owned delivery task.
|
||||
|
||||
async def try_delivery():
|
||||
async with jobs.state_lock:
|
||||
delivery = replies.claim()
|
||||
if delivery is None:
|
||||
return
|
||||
events = replies.prepare(
|
||||
delivery.delivery_id,
|
||||
lambda current_jobs: "\n\n".join(job.answer for job in current_jobs),
|
||||
)
|
||||
for event in events:
|
||||
async with jobs.state_lock:
|
||||
if not replies.accepts_output(delivery.delivery_id):
|
||||
replies.settle(delivery.delivery_id, "skipped")
|
||||
return
|
||||
if turns.accept_event(event):
|
||||
await send_public_event(event)
|
||||
# Generation has completed. The speaking floor is still leased.
|
||||
# Validate whole-message adapter feedback before calling:
|
||||
# replies.settle(delivery.delivery_id, "completed")
|
||||
```
|
||||
|
||||
Serialize enqueue, claim, preparation, coverage changes, and output handoff with the same `JobRunner.state_lock` used by foreground turns and job commits. `ReplyQueue` also protects its own controls with a lock so two claims cannot acquire the floor. `prepare()` reads copied **current accepted artifacts**, rather than cached text captured at enqueue, and emits the shared `started`/`text`/`completed` public reply events with origin and job identity. It does not execute an LLM, append history, or schedule TTS. The application chooses intentional public text, history insertion, delivery tasks and transport.
|
||||
|
||||
`set_foreground(True)` holds the gate through foreground generation and draining, including pending user requests; release it when that work has settled. `observe_client(ClientActivity(...))` accepts only the owning session and increasing client sequence, with strict boolean `recording`, `playback`, and `muted` observations. Identified playback must belong to the Flow; a stale stop for a different reply cannot clear newer playback. Null `delivery_id` accommodates a local cached acknowledgment. Recording, playback, mute, a running turn, session closure or an already active delivery blocks `claim()`. If recording/mute or a new user request arrives during an announcement, the adapter must interrupt and stop that active delivery, then update activity and wake scheduling.
|
||||
|
||||
`claim()` atomically changes one eligible delivery from `pending` to `scheduled`. Before preparation and **each** handoff, recheck relevance with `accepts_output(delivery_id)` and claim text events with `turns.accept_event(event)`. Identity/version changes, missing jobs, non-successful work, ended sessions, already-covered updates and application `is_relevant(job)` policy invalidate announcements. The callback must be side-effect free; use it for application cancellation/supersession beyond the existing job contract. Wake the delivery task when activity or relevance changes. Relevance skips never remove retained artifacts.
|
||||
|
||||
After a requested summary has actually been handed off, `mark_covered(CoveredUpdate.from_job(job))` records the accepted version and skips matching pending announcements. Merely requesting a summary is not delivery. A later job version remains eligible. Do not mark all context jobs covered by a status response or unrelated conversation.
|
||||
|
||||
`settle(delivery_id, status, reason=...)` releases only the active delivery, once. Its terminal status is `completed`, `interrupted`, `skipped`, or `failed`; generation completion alone does **not** settle delivery. Validate feedback session, turn, input revision and delivery identity at the adapter boundary, and accept completion only after output generation has settled. Whole-message adapter completion is a scheduling observation, not proof of audible words. On interruption, failure or invalidation, stop local playback and in-flight adapter delivery, fence remaining public output, and leave the result available. Interrupted/failed announcements are not automatically replayed; users can request a fresh summary. TTS failure does not change job success.
|
||||
|
||||
Call `close()` before cancelling and draining your application-owned delivery tasks. It fences pending/active deliveries while retaining job outputs. Snapshot data includes floor observations and delivery records, not a timer, media connection or resumable lease. Restoration does not restart delivery; reconcile active records through application recovery policy. Segment-level receipts, exact spoken context, automatic job cancellation/replacement, distributed scheduling and playback timeouts remain separate features. A client that never reports completion conservatively retains the floor until interrupted or closed; applications can add an explicit timeout policy.
|
||||
|
||||
@@ -741,3 +741,63 @@ runner.interrupt(
|
||||
이 단계는 백그라운드 응답 queue, 발언권 스케줄링, 재생 receipt, 정확한 단어 정렬, speculative generation, job pause/resume를 제공하지 않습니다. snapshot에는 기록만 있고 lock, 취소 신호, 실행 중인 작업은 없습니다. 작업을 재개하거나 delivery를 다시 재생하지 않습니다. 새 추적 live turn 전에 상태를 복원하세요. wrapper는 `from_checkpoint` 및 `restore_from_state_id`를 의도적으로 거부합니다. 영속적인 조정과 최종 추적 기록 저장은 별도 작업입니다. 기존 checkpoint 워크플로에는 안정적인 Flow API를 계속 사용하세요.
|
||||
|
||||
추적 실행 중에는 현재 복원된 상태가 기준이 됩니다. 자동 세션 재로드를 억제하여 등록된 turn/reply 기록이나 승인된 백그라운드 job 업데이트가 사라지지 않도록 합니다. 기본 Flow 턴 API의 기존 영속 상태 재로드 동작은 유지됩니다. 종료 이벤트 이후 iterator를 닫아도 완료된 응답을 중단하지 않습니다. 실패 이후 중단 요청이 도착해도 실패 이벤트는 유효한 종료 수명 주기 신호로 처리됩니다.
|
||||
|
||||
## 실험적 백그라운드 응답 큐
|
||||
|
||||
`crewai.experimental.flow_replies`는 턴/응답 계약 위에 `ReplyQueueState`, `ReplyQueue`, `DeliveryRecord`, `DeliveryFloor`, `CoveredUpdate`, `ClientActivity`를 추가합니다. 상태에 `ReplyQueueState`와 `JobState[MyJob]`를 조합하고 live Flow마다 `TurnRunner` **하나**와 `ReplyQueue` **하나**를 만드세요. 기존 Flow는 선택적으로 사용하며 job 실행기와 안정적인 대화 API는 바뀌지 않습니다.
|
||||
|
||||
성공한 job 결과는 안내가 기다리는 동안에도 즉시 상태에 있습니다. `enqueue(job)`은 현재 커밋된 `completed` job과 기존 원래 턴만 허용합니다. 대기 delivery는 job의 revision, attempt, 승인된 업데이트 순서를 저장하고 중복 알림을 같은 `delivery_id`로 합칩니다. 이 ID가 이후 공개 `ReplyRecord`의 ID이기도 합니다. job 완료마다 출처를 추적할 수 있는 안내 하나를 만들며 중간 단계 진행은 발화 큐를 만들지 않습니다.
|
||||
|
||||
```python
|
||||
from crewai.experimental.flow_jobs import JobState
|
||||
from crewai.experimental.flow_turns import TurnRunner
|
||||
from crewai.experimental.flow_replies import (
|
||||
ClientActivity, CoveredUpdate, ReplyQueue, ReplyQueueState,
|
||||
)
|
||||
|
||||
class WorkChatState(ReplyQueueState, JobState[MyJob]):
|
||||
pass
|
||||
|
||||
# Your conversational Flow uses WorkChatState; jobs is its JobRunner.
|
||||
turns = TurnRunner(flow)
|
||||
replies = ReplyQueue(turns, is_relevant=lambda job: job.job_id not in cancelled_job_ids)
|
||||
|
||||
async def on_job_snapshot(snapshot):
|
||||
async with jobs.state_lock:
|
||||
for job in flow.state.jobs.values():
|
||||
if job.status == "completed":
|
||||
replies.enqueue(job)
|
||||
# Publish snapshots and wake your application-owned delivery task.
|
||||
|
||||
async def try_delivery():
|
||||
async with jobs.state_lock:
|
||||
delivery = replies.claim()
|
||||
if delivery is None:
|
||||
return
|
||||
events = replies.prepare(
|
||||
delivery.delivery_id,
|
||||
lambda current_jobs: "\n\n".join(job.answer for job in current_jobs),
|
||||
)
|
||||
for event in events:
|
||||
async with jobs.state_lock:
|
||||
if not replies.accepts_output(delivery.delivery_id):
|
||||
replies.settle(delivery.delivery_id, "skipped")
|
||||
return
|
||||
if turns.accept_event(event):
|
||||
await send_public_event(event)
|
||||
# Generation has completed. The speaking floor is still leased.
|
||||
# Validate whole-message adapter feedback before calling:
|
||||
# replies.settle(delivery.delivery_id, "completed")
|
||||
```
|
||||
|
||||
enqueue, claim, 준비, coverage 변경, 출력 전달을 foreground 턴 및 job 커밋과 같은 `JobRunner.state_lock`으로 직렬화하세요. `ReplyQueue`는 자체 제어도 lock으로 보호하므로 두 claim이 동시에 발언권을 얻을 수 없습니다. `prepare()`는 enqueue 당시 저장한 텍스트가 아니라 복사한 **현재 승인된 결과**를 읽고 원래 턴과 job ID를 포함한 공통 `started`/`text`/`completed` 이벤트를 만듭니다. LLM 실행, history 삽입, TTS 예약은 하지 않습니다. 공개 텍스트, history, delivery task, transport는 애플리케이션이 선택합니다.
|
||||
|
||||
`set_foreground(True)`는 대기 중인 사용자 요청을 포함하여 foreground 생성과 종료 대기 동안 gate를 유지합니다. 작업이 끝난 뒤 해제하세요. `observe_client(ClientActivity(...))`는 소유 세션과 증가하는 client 순서, 엄격한 boolean `recording`, `playback`, `muted`만 허용합니다. ID가 있는 playback은 이 Flow 소유여야 하며 다른 응답의 오래된 stop은 새 playback을 해제할 수 없습니다. null `delivery_id`는 로컬 캐시 acknowledgment를 허용합니다. 녹음, playback, mute, 실행 중인 턴, 세션 종료, 이미 활성인 delivery는 `claim()`을 막습니다. 안내 중 녹음/mute 또는 새 사용자 요청이 오면 adapter가 활성 delivery를 중단하고 멈춘 뒤 activity를 갱신하고 스케줄러를 깨워야 합니다.
|
||||
|
||||
`claim()`은 적합한 delivery 하나를 원자적으로 `pending`에서 `scheduled`로 바꿉니다. 준비 전과 **모든** 전달 전에 `accepts_output(delivery_id)`로 relevance를 확인하고 `turns.accept_event(event)`로 텍스트 이벤트를 승인하세요. ID/버전 변경, 사라진 job, 성공하지 않은 작업, 끝난 세션, 이미 coverage된 업데이트, 애플리케이션 `is_relevant(job)` 정책은 안내를 무효화합니다. callback은 부수 효과가 없어야 합니다. 기존 job 계약 밖의 애플리케이션 취소/교체 정책에 사용하세요. activity 또는 relevance가 바뀌면 delivery task를 깨우세요. 건너뛰어도 보존한 결과는 삭제하지 않습니다.
|
||||
|
||||
요청한 요약이 실제 전달된 뒤 `mark_covered(CoveredUpdate.from_job(job))`은 승인된 버전을 기록하고 일치하는 pending 안내를 건너뜁니다. 요약 요청만으로 delivery가 되지 않습니다. 이후 job 버전은 계속 적합합니다. 상태 응답이나 관련 없는 대화로 context의 모든 job을 coverage 처리하지 마세요.
|
||||
|
||||
`settle(delivery_id, status, reason=...)`은 활성 delivery만 한 번 해제합니다. 최종 상태는 `completed`, `interrupted`, `skipped`, `failed`입니다. 생성 완료만으로 delivery가 완료되지는 **않습니다**. adapter 경계에서 feedback의 세션, 턴, input revision, delivery ID를 검증하고 출력 생성이 끝난 뒤에만 completion을 받으세요. 전체 메시지 adapter completion은 스케줄링 관측이며 들린 단어의 증거가 아닙니다. 중단, 실패, 무효화 시 로컬 playback과 진행 중인 adapter delivery를 멈추고 남은 공개 출력을 차단하며 결과를 보존하세요. 중단/실패 안내는 자동 재생하지 않고 사용자가 새 요약을 요청할 수 있습니다. TTS 실패는 job 성공을 바꾸지 않습니다.
|
||||
|
||||
애플리케이션 delivery task를 취소하고 기다리기 전에 `close()`를 호출하세요. pending/active delivery를 차단하고 job 결과를 보존합니다. snapshot에는 발언권 관측과 delivery 기록이 있고 timer, media 연결, 재개 가능한 lease는 없습니다. 복원은 delivery를 재시작하지 않습니다. 활성 기록은 애플리케이션 recovery 정책으로 조정하세요. segment receipt, 정확한 spoken context, 자동 job 취소/교체, 분산 스케줄링, playback timeout은 별도 기능입니다. completion을 보내지 않는 client는 보수적으로 중단 또는 종료까지 발언권을 유지합니다. 애플리케이션이 명시적인 timeout 정책을 추가할 수 있습니다.
|
||||
|
||||
@@ -746,3 +746,63 @@ Para combinar com trabalho em segundo plano, declare `class WorkChatState(TurnSt
|
||||
Esta etapa não oferece filas de respostas em segundo plano, agendamento da vez de falar, recibos de reprodução, alinhamento exato de palavras, geração especulativa ou pause/resume de jobs. Snapshots contêm registros, não locks, sinais de cancelamento ou execuções ativas. Não retomam trabalho nem repetem entregas. Restaure o estado antes de criar um novo turno acompanhado; este wrapper rejeita `from_checkpoint` e `restore_from_state_id` deliberadamente. Reconciliação durável e persistência dos registros finais de acompanhamento continuam sendo trabalhos separados. Use as APIs estáveis de Flow para workflows existentes de checkpoint.
|
||||
|
||||
Durante a execução acompanhada, o estado atual restaurado permanece como referência: o recarregamento automático da sessão é suprimido para não descartar registros de turn/reply admitidos nem atualizações aceitas de jobs em segundo plano. As APIs nativas de turno do Flow mantêm seu comportamento de recarregamento de persistência. Fechar o iterator após um evento terminal não interrompe a resposta concluída. Uma falha continua sendo um sinal terminal de ciclo de vida aceito se a interrupção chegar após a falha.
|
||||
|
||||
## Respostas experimentais em fila para trabalho em segundo plano
|
||||
|
||||
`crewai.experimental.flow_replies` adiciona `ReplyQueueState`, `ReplyQueue`, `DeliveryRecord`, `DeliveryFloor`, `CoveredUpdate` e `ClientActivity` ao contrato de turnos/respostas. Combine `ReplyQueueState` com `JobState[MyJob]` e crie **um** `TurnRunner` e **uma** `ReplyQueue` por Flow ativo. A integração é opcional; o executor de jobs e as APIs estáveis de conversa não mudam.
|
||||
|
||||
Um job bem-sucedido fica disponível imediatamente no estado, mesmo quando seu anúncio espera. `enqueue(job)` aceita apenas um job atual, já confirmado como `completed`, com turno de origem existente. A entrega pendente guarda revision, attempt e sequência de atualização aceita; notificações duplicadas usam o mesmo `delivery_id`. Esse ID também identifica o `ReplyRecord` público posterior. Cada conclusão de job recebe um anúncio atribuível; progresso intermediário de etapas não cria uma fila de fala.
|
||||
|
||||
```python
|
||||
from crewai.experimental.flow_jobs import JobState
|
||||
from crewai.experimental.flow_turns import TurnRunner
|
||||
from crewai.experimental.flow_replies import (
|
||||
ClientActivity, CoveredUpdate, ReplyQueue, ReplyQueueState,
|
||||
)
|
||||
|
||||
class WorkChatState(ReplyQueueState, JobState[MyJob]):
|
||||
pass
|
||||
|
||||
# Your conversational Flow uses WorkChatState; jobs is its JobRunner.
|
||||
turns = TurnRunner(flow)
|
||||
replies = ReplyQueue(turns, is_relevant=lambda job: job.job_id not in cancelled_job_ids)
|
||||
|
||||
async def on_job_snapshot(snapshot):
|
||||
async with jobs.state_lock:
|
||||
for job in flow.state.jobs.values():
|
||||
if job.status == "completed":
|
||||
replies.enqueue(job)
|
||||
# Publish snapshots and wake your application-owned delivery task.
|
||||
|
||||
async def try_delivery():
|
||||
async with jobs.state_lock:
|
||||
delivery = replies.claim()
|
||||
if delivery is None:
|
||||
return
|
||||
events = replies.prepare(
|
||||
delivery.delivery_id,
|
||||
lambda current_jobs: "\n\n".join(job.answer for job in current_jobs),
|
||||
)
|
||||
for event in events:
|
||||
async with jobs.state_lock:
|
||||
if not replies.accepts_output(delivery.delivery_id):
|
||||
replies.settle(delivery.delivery_id, "skipped")
|
||||
return
|
||||
if turns.accept_event(event):
|
||||
await send_public_event(event)
|
||||
# Generation has completed. The speaking floor is still leased.
|
||||
# Validate whole-message adapter feedback before calling:
|
||||
# replies.settle(delivery.delivery_id, "completed")
|
||||
```
|
||||
|
||||
Serialize enqueue, claim, preparação, alterações de coverage e envio de saída com o mesmo `JobRunner.state_lock` usado por turnos e commits de jobs. `ReplyQueue` protege seus controles com outro lock para que dois claims não adquiram a vez de falar. `prepare()` lê cópias dos **resultados atuais aceitos**, em vez de texto guardado no enqueue, e emite os eventos públicos compartilhados `started`/`text`/`completed` com origem e identidade do job. Não executa LLM, adiciona histórico ou agenda TTS. A aplicação escolhe texto público, histórico, tarefas de entrega e transporte.
|
||||
|
||||
`set_foreground(True)` mantém o bloqueio durante geração e encerramento em primeiro plano, incluindo pedidos pendentes do usuário; libere após o trabalho terminar. `observe_client(ClientActivity(...))` aceita apenas a sessão dona e sequência crescente do client, com booleanos estritos `recording`, `playback` e `muted`. Playback identificado deve pertencer ao Flow; um stop antigo de outra resposta não pode liberar playback mais recente. `delivery_id` nulo permite acknowledgment local em cache. Gravação, playback, mute, turno em execução, sessão encerrada ou entrega já ativa bloqueiam `claim()`. Se gravação/mute ou um novo pedido chegar durante um anúncio, o adapter deve interromper e parar a entrega ativa, atualizar atividade e acordar o agendamento.
|
||||
|
||||
`claim()` muda atomicamente uma entrega elegível de `pending` para `scheduled`. Antes da preparação e de **cada** envio, confira relevância com `accepts_output(delivery_id)` e aceite eventos de texto com `turns.accept_event(event)`. Mudanças de identidade/versão, jobs ausentes, trabalho sem sucesso, sessões encerradas, atualizações já cobertas e a política `is_relevant(job)` invalidam anúncios. O callback deve ser livre de efeitos colaterais; use-o para cancelamento/substituição da aplicação além do contrato atual de jobs. Acorde a tarefa quando atividade ou relevância mudar. Pular anúncios nunca remove resultados retidos.
|
||||
|
||||
Depois que um resumo solicitado for realmente enviado, `mark_covered(CoveredUpdate.from_job(job))` registra a versão aceita e pula anúncios pendentes correspondentes. Apenas pedir um resumo não é entrega. Uma versão posterior do job continua elegível. Não marque todos os jobs do contexto como cobertos por uma resposta de status ou conversa não relacionada.
|
||||
|
||||
`settle(delivery_id, status, reason=...)` libera apenas a entrega ativa, uma vez. Os estados finais são `completed`, `interrupted`, `skipped` e `failed`; terminar a geração **não** conclui a entrega. Valide sessão, turno, input revision e identidade no adapter e aceite conclusão apenas após a geração da saída terminar. Conclusão de mensagem inteira no adapter é uma observação de agendamento, não prova de palavras audíveis. Em interrupção, falha ou invalidação, pare playback local e entrega em andamento, bloqueie saída pública restante e mantenha o resultado disponível. Anúncios interrompidos/falhos não repetem automaticamente; usuários podem pedir um novo resumo. Falha no TTS não altera sucesso do job.
|
||||
|
||||
Chame `close()` antes de cancelar e aguardar tarefas de entrega da aplicação. Isso bloqueia entregas pendentes/ativas e mantém resultados dos jobs. Snapshot inclui observações da vez de falar e registros de entrega, não timer, conexão de mídia ou lease retomável. Restaurar não reinicia entrega; reconcilie registros ativos pela política de recuperação da aplicação. Recibos de segmentos, contexto falado exato, cancelamento/substituição automática de jobs, agendamento distribuído e timeout de playback são funcionalidades separadas. Um client que nunca informa conclusão mantém conservadoramente a vez de falar até interrupção ou fechamento; a aplicação pode adicionar uma política explícita de timeout.
|
||||
|
||||
@@ -0,0 +1,449 @@
|
||||
"""Experimental background reply queue over accepted job and turn state.
|
||||
|
||||
Call mutations through the application's serialized state path (for example,
|
||||
JobRunner.state_lock). This owns reply scheduling, not jobs, TTS or live tasks.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
from threading import RLock
|
||||
from time import perf_counter
|
||||
from typing import Any, Literal, cast
|
||||
from uuid import uuid4
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
from crewai.experimental.flow_jobs import JobRecord, JobState, _timestamp
|
||||
from crewai.experimental.flow_turns import (
|
||||
ReplyEvent,
|
||||
ReplyRecord,
|
||||
TurnRunner,
|
||||
TurnState,
|
||||
)
|
||||
|
||||
|
||||
class CoveredUpdate(BaseModel):
|
||||
"""The accepted job version included in an announcement or manual summary."""
|
||||
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
job_id: str = Field(min_length=1)
|
||||
revision: int = Field(gt=0, strict=True)
|
||||
attempt: int = Field(gt=0, strict=True)
|
||||
seq: int = Field(ge=0, strict=True)
|
||||
|
||||
@classmethod
|
||||
def from_job(cls, job: JobRecord) -> CoveredUpdate:
|
||||
return cls(
|
||||
job_id=job.job_id,
|
||||
revision=job.revision,
|
||||
attempt=job.attempt,
|
||||
seq=job.last_update_seq,
|
||||
)
|
||||
|
||||
|
||||
class DeliveryRecord(BaseModel):
|
||||
"""One queued public reply; delivery_id is also its ReplyRecord identity."""
|
||||
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
session_id: str = Field(min_length=1)
|
||||
delivery_id: str = Field(default_factory=lambda: str(uuid4()), min_length=1)
|
||||
turn_id: str = Field(min_length=1)
|
||||
input_revision: int = Field(gt=0, strict=True)
|
||||
covered_updates: list[CoveredUpdate]
|
||||
status: Literal[
|
||||
"pending", "scheduled", "completed", "interrupted", "skipped", "failed"
|
||||
] = "pending"
|
||||
reason: str = ""
|
||||
created_at: str = Field(default_factory=_timestamp)
|
||||
updated_at: str = Field(default_factory=_timestamp)
|
||||
scheduled_at: str | None = None
|
||||
settled_at: str | None = None
|
||||
|
||||
|
||||
class DeliveryFloor(BaseModel):
|
||||
"""Serializable gate observations, never media handles or resumable leases."""
|
||||
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
foreground: bool = False
|
||||
recording: bool = False
|
||||
playback: bool = False
|
||||
muted: bool = False
|
||||
closed: bool = False
|
||||
client_seq: int = 0
|
||||
playback_delivery_id: str | None = None
|
||||
active_delivery_id: str | None = None
|
||||
|
||||
|
||||
class ClientActivity(BaseModel):
|
||||
"""Session-owned aggregate client activity; seq must increase on that client."""
|
||||
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
session_id: str = Field(min_length=1)
|
||||
seq: int = Field(gt=0, strict=True)
|
||||
recording: bool = Field(strict=True)
|
||||
playback: bool = Field(strict=True)
|
||||
muted: bool = Field(strict=True)
|
||||
delivery_id: str | None = None
|
||||
|
||||
|
||||
class ReplyQueueState(TurnState):
|
||||
"""Compose with JobState[MyJob]; generation and delivery remain separate."""
|
||||
|
||||
deliveries: dict[str, DeliveryRecord] = Field(default_factory=dict)
|
||||
covered_updates: dict[str, CoveredUpdate] = Field(default_factory=dict)
|
||||
delivery_floor: DeliveryFloor = Field(default_factory=DeliveryFloor)
|
||||
delivery_sequence: int = 0
|
||||
|
||||
|
||||
class ReplyQueue:
|
||||
"""One queue and speaking floor per live conversational Flow.
|
||||
|
||||
Jobs commit before enqueue. Claim and prepare recheck accepted job versions;
|
||||
prepare calls the application's text builder with current copied records.
|
||||
Supply is_relevant for application cancellation/supersession policy. Calls
|
||||
never remove outputs or change job success. Restore does not restart delivery.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
turns: TurnRunner,
|
||||
*,
|
||||
is_relevant: Callable[[JobRecord], bool] | None = None,
|
||||
) -> None:
|
||||
self.turns = turns
|
||||
self.is_relevant = is_relevant or (lambda job: True)
|
||||
self._lock = RLock()
|
||||
if not (
|
||||
isinstance(turns.state, ReplyQueueState)
|
||||
and isinstance(turns.state, JobState)
|
||||
):
|
||||
raise TypeError(
|
||||
"ReplyQueue requires ReplyQueueState composed with JobState"
|
||||
)
|
||||
|
||||
@property
|
||||
def state(self) -> ReplyQueueState:
|
||||
state = self.turns.state
|
||||
if not (isinstance(state, ReplyQueueState) and isinstance(state, JobState)):
|
||||
raise TypeError("Restored state must include ReplyQueueState and JobState")
|
||||
return state
|
||||
|
||||
def _change(self, record: DeliveryRecord, status: Any, reason: str = "") -> None:
|
||||
record.status, record.reason = status, reason
|
||||
record.updated_at = _timestamp()
|
||||
self.state.delivery_sequence += 1
|
||||
if status == "scheduled":
|
||||
record.scheduled_at = record.updated_at
|
||||
if status not in {"pending", "scheduled"}:
|
||||
record.settled_at = record.updated_at
|
||||
if self.state.delivery_floor.active_delivery_id == record.delivery_id:
|
||||
self.state.delivery_floor.active_delivery_id = None
|
||||
|
||||
def _current(self, coverage: CoveredUpdate) -> JobRecord | None:
|
||||
job = cast(JobState[JobRecord], self.state).jobs.get(coverage.job_id)
|
||||
if (
|
||||
job is None
|
||||
or job.session_id != self.state.id
|
||||
or job.status != "completed"
|
||||
or (job.revision, job.attempt, job.last_update_seq)
|
||||
!= (coverage.revision, coverage.attempt, coverage.seq)
|
||||
or not self.is_relevant(job.model_copy(deep=True))
|
||||
):
|
||||
return None
|
||||
return job
|
||||
|
||||
def _covered(self, coverage: CoveredUpdate) -> bool:
|
||||
accepted = self.state.covered_updates.get(coverage.job_id)
|
||||
return bool(
|
||||
accepted
|
||||
and (accepted.revision, accepted.attempt)
|
||||
== (coverage.revision, coverage.attempt)
|
||||
and accepted.seq >= coverage.seq
|
||||
)
|
||||
|
||||
def enqueue(self, job: JobRecord) -> str | None:
|
||||
"""Queue one accepted completion; duplicates coalesce by job version.
|
||||
|
||||
Only terminal completions are announced, so noisy stage progress never
|
||||
becomes a speaking queue. Multiple jobs retain separate attributable replies.
|
||||
"""
|
||||
with self._lock:
|
||||
coverage = CoveredUpdate.from_job(job)
|
||||
current = self._current(coverage)
|
||||
turn = self.state.turns.get(job.origin_turn_id)
|
||||
if (
|
||||
self.state.ended
|
||||
or self.state.delivery_floor.closed
|
||||
or job.session_id != self.state.id
|
||||
or current is None
|
||||
or current.origin_turn_id != job.origin_turn_id
|
||||
or turn is None
|
||||
or turn.session_id != self.state.id
|
||||
or self._covered(coverage)
|
||||
):
|
||||
return None
|
||||
for record in self.state.deliveries.values():
|
||||
if coverage in record.covered_updates:
|
||||
return record.delivery_id
|
||||
record = DeliveryRecord(
|
||||
session_id=self.state.id,
|
||||
turn_id=turn.turn_id,
|
||||
input_revision=turn.input_revision,
|
||||
covered_updates=[coverage],
|
||||
)
|
||||
self.state.deliveries[record.delivery_id] = record
|
||||
self.state.delivery_sequence += 1
|
||||
return record.delivery_id
|
||||
|
||||
def set_foreground(self, busy: bool) -> None:
|
||||
"""Server-owned activity; keep true through foreground generation/drain."""
|
||||
with self._lock:
|
||||
self.state.delivery_floor.foreground = busy
|
||||
|
||||
def observe_client(self, update: ClientActivity) -> bool:
|
||||
"""Accept ordered session activity, not proof of physical audible output.
|
||||
|
||||
A null delivery_id permits local cached audio. Identified playback must
|
||||
belong to this Flow; a stale stop cannot release newer identified playback.
|
||||
"""
|
||||
with self._lock:
|
||||
floor = self.state.delivery_floor
|
||||
if (
|
||||
floor.closed
|
||||
or update.session_id != self.state.id
|
||||
or update.seq <= floor.client_seq
|
||||
):
|
||||
return False
|
||||
if update.delivery_id is not None:
|
||||
reply = self.state.replies.get(update.delivery_id)
|
||||
if reply is None or reply.session_id != self.state.id:
|
||||
return False
|
||||
if (
|
||||
not update.playback
|
||||
and floor.playback
|
||||
and floor.playback_delivery_id != update.delivery_id
|
||||
):
|
||||
return False
|
||||
floor.client_seq = update.seq
|
||||
floor.recording, floor.playback, floor.muted = (
|
||||
update.recording,
|
||||
update.playback,
|
||||
update.muted,
|
||||
)
|
||||
floor.playback_delivery_id = update.delivery_id if update.playback else None
|
||||
return True
|
||||
|
||||
def _eligible(self, record: DeliveryRecord) -> bool:
|
||||
turn = self.state.turns.get(record.turn_id)
|
||||
return bool(
|
||||
not self.state.ended
|
||||
and record.session_id == self.state.id
|
||||
and turn is not None
|
||||
and (turn.session_id, turn.input_revision)
|
||||
== (record.session_id, record.input_revision)
|
||||
and record.covered_updates
|
||||
and all(
|
||||
self._current(c) is not None and not self._covered(c)
|
||||
for c in record.covered_updates
|
||||
)
|
||||
)
|
||||
|
||||
def claim(self) -> DeliveryRecord | None:
|
||||
"""Atomically acquire the floor for at most one still-relevant completion."""
|
||||
with self._lock:
|
||||
floor = self.state.delivery_floor
|
||||
for record in self.state.deliveries.values():
|
||||
if record.status == "pending" and not self._eligible(record):
|
||||
self._change(
|
||||
record, "skipped", "No longer relevant or already covered."
|
||||
)
|
||||
if (
|
||||
floor.closed
|
||||
or floor.foreground
|
||||
or floor.recording
|
||||
or floor.playback
|
||||
or floor.muted
|
||||
or floor.active_delivery_id is not None
|
||||
or any(t.status == "running" for t in self.state.turns.values())
|
||||
):
|
||||
return None
|
||||
for record in self.state.deliveries.values():
|
||||
if record.status == "pending":
|
||||
floor.active_delivery_id = record.delivery_id
|
||||
self._change(record, "scheduled")
|
||||
return record.model_copy(deep=True)
|
||||
return None
|
||||
|
||||
def prepare(
|
||||
self, delivery_id: str, build_text: Callable[[list[JobRecord]], str]
|
||||
) -> list[ReplyEvent]:
|
||||
"""Build public text from current artifacts after claim, before handoff.
|
||||
|
||||
Records/events reuse the TurnRunner contract. History insertion is an
|
||||
application choice; this neither invokes an LLM nor claims audio completion.
|
||||
"""
|
||||
with self._lock:
|
||||
if delivery_id in self.state.replies:
|
||||
return [] # Preparation is single-use, never duplicate public output.
|
||||
record = self.state.deliveries.get(delivery_id)
|
||||
floor = self.state.delivery_floor
|
||||
if (
|
||||
record is None
|
||||
or record.status != "scheduled"
|
||||
or floor.active_delivery_id != delivery_id
|
||||
or not self._eligible(record)
|
||||
):
|
||||
if record is not None and record.status == "scheduled":
|
||||
self._change(record, "skipped", "Invalidated before preparation.")
|
||||
return []
|
||||
if (
|
||||
floor.closed
|
||||
or floor.foreground
|
||||
or floor.recording
|
||||
or floor.playback
|
||||
or floor.muted
|
||||
):
|
||||
self._change(record, "interrupted", "The speaking floor became busy.")
|
||||
return []
|
||||
started = perf_counter()
|
||||
try:
|
||||
current = [self._current(c) for c in record.covered_updates]
|
||||
text = build_text(
|
||||
[job.model_copy(deep=True) for job in current if job is not None]
|
||||
)
|
||||
if not isinstance(text, str) or not text.strip():
|
||||
raise ValueError("An announcement requires non-empty public text")
|
||||
except Exception:
|
||||
self._change(record, "failed", "Reply preparation failed.")
|
||||
raise
|
||||
coverage = record.covered_updates[0]
|
||||
reply = ReplyRecord(
|
||||
session_id=record.session_id,
|
||||
turn_id=record.turn_id,
|
||||
input_revision=record.input_revision,
|
||||
delivery_id=delivery_id,
|
||||
kind="progress",
|
||||
status="completed",
|
||||
segments=[text],
|
||||
text=text,
|
||||
last_event_seq=3,
|
||||
job_id=coverage.job_id,
|
||||
job_revision=coverage.revision,
|
||||
job_attempt=coverage.attempt,
|
||||
)
|
||||
self.state.replies[delivery_id] = reply
|
||||
details = dict(
|
||||
session_id=reply.session_id,
|
||||
turn_id=reply.turn_id,
|
||||
input_revision=reply.input_revision,
|
||||
delivery_id=delivery_id,
|
||||
kind=reply.kind,
|
||||
elapsed_ms=(perf_counter() - started) * 1000,
|
||||
job_id=reply.job_id,
|
||||
job_revision=reply.job_revision,
|
||||
job_attempt=reply.job_attempt,
|
||||
)
|
||||
return [
|
||||
ReplyEvent(type="started", seq=1, **details),
|
||||
ReplyEvent(type="text", seq=2, text=text, segment_id=1, **details),
|
||||
ReplyEvent(type="completed", seq=3, text=text, **details),
|
||||
]
|
||||
|
||||
def mark_covered(self, coverage: CoveredUpdate) -> bool:
|
||||
"""Call only after an explicit summary of this accepted version is delivered."""
|
||||
with self._lock:
|
||||
if self._current(coverage) is None or self.state.delivery_floor.closed:
|
||||
return False
|
||||
self.state.covered_updates[coverage.job_id] = coverage.model_copy(deep=True)
|
||||
for record in self.state.deliveries.values():
|
||||
if record.status == "pending" and any(
|
||||
self._covered(c) for c in record.covered_updates
|
||||
):
|
||||
self._change(
|
||||
record, "skipped", "Already covered by a requested summary."
|
||||
)
|
||||
return True
|
||||
|
||||
def settle(
|
||||
self,
|
||||
delivery_id: str,
|
||||
status: Literal["completed", "interrupted", "skipped", "failed"],
|
||||
*,
|
||||
reason: str = "",
|
||||
) -> bool:
|
||||
"""Settle the active delivery once; adapters must validate feedback identity.
|
||||
|
||||
completed is a whole-message adapter observation, not a speaker/word
|
||||
receipt. Interrupted/failed announcements stay inspectable without retry.
|
||||
"""
|
||||
with self._lock:
|
||||
record = self.state.deliveries.get(delivery_id)
|
||||
if (
|
||||
record is None
|
||||
or record.status != "scheduled"
|
||||
or self.state.delivery_floor.active_delivery_id != delivery_id
|
||||
or status not in {"completed", "interrupted", "skipped", "failed"}
|
||||
):
|
||||
return False
|
||||
reply = self.state.replies.get(delivery_id)
|
||||
if status == "completed" and reply is not None and reply.interrupted:
|
||||
status, reason = "interrupted", "Public reply interrupted."
|
||||
elif status == "completed" and not self._eligible(record):
|
||||
status, reason = "skipped", "Invalidated before delivery completion."
|
||||
if status == "completed":
|
||||
for coverage in record.covered_updates:
|
||||
self.state.covered_updates[coverage.job_id] = coverage.model_copy(
|
||||
deep=True
|
||||
)
|
||||
else:
|
||||
self.turns.interrupt(
|
||||
session_id=record.session_id,
|
||||
turn_id=record.turn_id,
|
||||
input_revision=record.input_revision,
|
||||
delivery_id=delivery_id,
|
||||
)
|
||||
self._change(record, status, reason)
|
||||
return True
|
||||
|
||||
def accepts_output(self, delivery_id: str) -> bool:
|
||||
"""Recheck after queues/provider waits, immediately before transport handoff."""
|
||||
with self._lock:
|
||||
record = self.state.deliveries.get(delivery_id)
|
||||
reply = self.state.replies.get(delivery_id)
|
||||
return bool(
|
||||
record
|
||||
and reply is not None
|
||||
and not reply.interrupted
|
||||
and reply.status != "failed"
|
||||
and record.status == "scheduled"
|
||||
and self.state.delivery_floor.active_delivery_id == delivery_id
|
||||
and not self.state.delivery_floor.closed
|
||||
and not self.state.delivery_floor.foreground
|
||||
and not self.state.delivery_floor.recording
|
||||
and not self.state.delivery_floor.muted
|
||||
and self._eligible(record)
|
||||
)
|
||||
|
||||
def close(self) -> None:
|
||||
"""Fence pending/active delivery; retained job results are untouched."""
|
||||
with self._lock:
|
||||
self.state.delivery_floor.closed = True
|
||||
for record in self.state.deliveries.values():
|
||||
if record.status == "scheduled":
|
||||
self.settle(
|
||||
record.delivery_id, "interrupted", reason="Session closed."
|
||||
)
|
||||
elif record.status == "pending":
|
||||
self._change(record, "skipped", "Session closed.")
|
||||
|
||||
def snapshot(self) -> dict[str, Any]:
|
||||
with self._lock:
|
||||
return dict(
|
||||
session_id=self.state.id,
|
||||
seq=self.state.delivery_sequence,
|
||||
deliveries=[
|
||||
r.model_dump(mode="json") for r in self.state.deliveries.values()
|
||||
],
|
||||
floor=self.state.delivery_floor.model_dump(mode="json"),
|
||||
)
|
||||
@@ -0,0 +1,294 @@
|
||||
"""Behavior checks for queued delivery over accepted artifacts and public replies."""
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from typing import ClassVar
|
||||
|
||||
import pytest
|
||||
|
||||
from crewai.experimental.flow_jobs import JobRecord, JobState, JobUpdate, add_job, commit_job_update
|
||||
from crewai.experimental.flow_replies import ClientActivity, CoveredUpdate, ReplyQueue, ReplyQueueState
|
||||
from crewai.experimental.flow_turns import TurnRecord, TurnRunner, TurnState
|
||||
from crewai.flow import Flow, listen
|
||||
|
||||
|
||||
class Result(JobRecord):
|
||||
output_fields: ClassVar[frozenset[str]] = frozenset({"answer"})
|
||||
answer: str = ""
|
||||
|
||||
|
||||
class State(ReplyQueueState, JobState[Result]):
|
||||
pass
|
||||
|
||||
|
||||
class Chat(Flow[State]):
|
||||
conversational = True
|
||||
|
||||
def route_turn(self, context):
|
||||
return "ack"
|
||||
|
||||
@listen("ack")
|
||||
def respond(self):
|
||||
return "Accepted."
|
||||
|
||||
|
||||
def setup():
|
||||
flow = Chat(suppress_flow_events=True)
|
||||
turns = TurnRunner(flow)
|
||||
return flow, turns, ReplyQueue(turns)
|
||||
|
||||
|
||||
def complete(flow, name="one"):
|
||||
state = flow.state
|
||||
state.turns[name] = TurnRecord(session_id=state.id, turn_id=name, status="completed")
|
||||
job = Result(session_id=state.id, origin_turn_id=name, job_id=name)
|
||||
assert add_job(state, job)
|
||||
for seq, kind, details in [(1, "started", {}), (2, "stage_started", {"stage": "result"}),
|
||||
(3, "stage_completed", {"stage": "result", "outputs": {"answer": f"Result {name}"}}),
|
||||
(4, "completed", {})]:
|
||||
assert commit_job_update(state, JobUpdate(session_id=state.id, job_id=name, revision=1,
|
||||
attempt=1, seq=seq, kind=kind, **details))
|
||||
return job
|
||||
|
||||
|
||||
def activity(flow, seq, **kwargs):
|
||||
return ClientActivity(session_id=flow.state.id, seq=seq, recording=False, playback=False,
|
||||
muted=False, **kwargs)
|
||||
|
||||
|
||||
def test_two_completions_retained_while_gate_held_then_only_one_claims_floor():
|
||||
flow, turns, queue = setup()
|
||||
queue.set_foreground(True)
|
||||
jobs = [complete(flow, name) for name in ("one", "two")]
|
||||
ids = [queue.enqueue(job) for job in jobs]
|
||||
assert all(ids) and len(set(ids)) == 2
|
||||
assert queue.claim() is None
|
||||
assert [job.answer for job in flow.state.jobs.values()] == ["Result one", "Result two"]
|
||||
assert all(r.status == "pending" for r in flow.state.deliveries.values())
|
||||
queue.set_foreground(False)
|
||||
with ThreadPoolExecutor(2) as pool:
|
||||
claims = list(pool.map(lambda _: queue.claim(), range(2)))
|
||||
assert sum(c is not None for c in claims) == 1
|
||||
first = next(c for c in claims if c is not None)
|
||||
events = queue.prepare(first.delivery_id, lambda current: current[0].answer)
|
||||
assert [e.type for e in events] == ["started", "text", "completed"]
|
||||
assert {e.delivery_id for e in events} == {first.delivery_id}
|
||||
assert {e.job_id for e in events} == {"one"}
|
||||
assert all(turns.accept_event(e) for e in events)
|
||||
assert queue.claim() is None # Generated text does not release the speaking floor.
|
||||
assert queue.prepare(first.delivery_id, lambda _: "duplicate") == []
|
||||
assert queue.settle(first.delivery_id, "completed")
|
||||
assert not queue.settle(first.delivery_id, "completed")
|
||||
assert queue.claim().delivery_id == ids[1]
|
||||
restored = State.model_validate_json(flow.state.model_dump_json())
|
||||
assert restored.jobs["one"].answer == "Result one"
|
||||
assert restored.replies[first.delivery_id].text == "Result one"
|
||||
assert restored.deliveries[first.delivery_id].status == "completed"
|
||||
|
||||
|
||||
def test_completion_must_commit_and_duplicates_coalesce_without_replay():
|
||||
flow, _, queue = setup()
|
||||
job = complete(flow)
|
||||
stale = job.model_copy(update={"last_update_seq": 3})
|
||||
assert queue.enqueue(stale) is None
|
||||
delivery_id = queue.enqueue(job)
|
||||
assert queue.enqueue(job.model_copy(deep=True)) == delivery_id
|
||||
assert len(queue.snapshot()["deliveries"]) == 1
|
||||
queue.claim()
|
||||
assert queue.settle(delivery_id, "interrupted")
|
||||
assert queue.enqueue(job) == delivery_id
|
||||
assert queue.claim() is None
|
||||
assert flow.state.jobs[job.job_id].status == "completed"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("change", ["revision", "attempt", "failed", "cancelled"])
|
||||
def test_invalidated_job_is_skipped_without_losing_artifacts(change):
|
||||
flow, turns, queue = setup()
|
||||
job = complete(flow)
|
||||
relevant = True
|
||||
queue.is_relevant = lambda _: relevant
|
||||
delivery_id = queue.enqueue(job)
|
||||
if change in {"revision", "attempt"}:
|
||||
setattr(job, change, 2)
|
||||
elif change == "failed":
|
||||
job.status = "failed"
|
||||
else:
|
||||
relevant = False # Application cancellation policy, not a new job status.
|
||||
assert queue.claim() is None
|
||||
assert flow.state.deliveries[delivery_id].status == "skipped"
|
||||
assert job.answer == "Result one"
|
||||
assert delivery_id not in turns.state.replies
|
||||
|
||||
|
||||
def test_recheck_after_claim_and_before_output_blocks_stale_completion():
|
||||
flow, turns, queue = setup()
|
||||
job = complete(flow)
|
||||
delivery_id = queue.enqueue(job)
|
||||
queue.claim()
|
||||
job.revision = 2
|
||||
assert queue.prepare(delivery_id, lambda _: pytest.fail("Stale builder invoked")) == []
|
||||
assert flow.state.deliveries[delivery_id].status == "skipped"
|
||||
assert not queue.accepts_output(delivery_id)
|
||||
newer = queue.enqueue(job)
|
||||
assert newer != delivery_id
|
||||
queue.claim()
|
||||
events = queue.prepare(newer, lambda current: current[0].answer)
|
||||
job.attempt = 2
|
||||
assert not queue.accepts_output(newer)
|
||||
assert queue.settle(newer, "completed")
|
||||
assert flow.state.deliveries[newer].status == "skipped"
|
||||
assert not turns.accept_event(events[1])
|
||||
|
||||
|
||||
def test_requested_summary_suppresses_pending_and_later_duplicate_announcement():
|
||||
flow, _, queue = setup()
|
||||
job = complete(flow)
|
||||
delivery_id = queue.enqueue(job)
|
||||
assert queue.mark_covered(CoveredUpdate.from_job(job))
|
||||
assert queue.claim() is None
|
||||
assert flow.state.deliveries[delivery_id].reason == "Already covered by a requested summary."
|
||||
assert queue.enqueue(job) is None
|
||||
job2 = complete(flow, "two")
|
||||
assert queue.mark_covered(CoveredUpdate.from_job(job2))
|
||||
assert queue.enqueue(job2) is None # Summary may win before completion notification enqueue.
|
||||
|
||||
|
||||
def test_interruption_fences_queued_text_and_releases_floor_without_replay():
|
||||
flow, turns, queue = setup()
|
||||
job = complete(flow)
|
||||
delivery_id = queue.enqueue(job)
|
||||
queue.claim()
|
||||
events = queue.prepare(delivery_id, lambda current: current[0].answer)
|
||||
assert turns.accept_event(events[0])
|
||||
assert queue.settle(delivery_id, "interrupted")
|
||||
assert not turns.accept_event(events[1])
|
||||
assert not turns.accept_event(events[2])
|
||||
assert not queue.accepts_output(delivery_id)
|
||||
assert queue.claim() is None
|
||||
assert job.status == "completed" and job.answer == "Result one"
|
||||
assert flow.state.replies[delivery_id].status == "completed"
|
||||
assert flow.state.replies[delivery_id].interrupted
|
||||
|
||||
|
||||
def test_recording_mute_playback_and_foreground_gate_preserve_pending_reply():
|
||||
flow, turns, queue = setup()
|
||||
job = complete(flow)
|
||||
delivery_id = queue.enqueue(job)
|
||||
for seq, field in enumerate(("recording", "muted", "playback"), 1):
|
||||
data = dict(session_id=flow.state.id, seq=seq, recording=False, muted=False, playback=False)
|
||||
data[field] = True
|
||||
assert queue.observe_client(ClientActivity(**data))
|
||||
assert queue.claim() is None
|
||||
assert queue.observe_client(activity(flow, 4))
|
||||
stream = turns.stream_turn("Hi", turn_id="foreground")
|
||||
next(stream)
|
||||
assert queue.claim() is None # Even if adapter forgot its server-owned busy gate.
|
||||
stream.close()
|
||||
assert queue.claim().delivery_id == delivery_id
|
||||
|
||||
|
||||
def test_foreign_duplicate_and_stale_client_observations_cannot_release_floor():
|
||||
flow, turns, queue = setup()
|
||||
job = complete(flow)
|
||||
delivery_id = queue.enqueue(job)
|
||||
queue.claim()
|
||||
queue.prepare(delivery_id, lambda current: current[0].answer)
|
||||
assert queue.observe_client(ClientActivity(session_id=flow.state.id, seq=1,
|
||||
recording=False, playback=True, muted=False, delivery_id=delivery_id))
|
||||
assert not queue.observe_client(activity(flow, 1, delivery_id=delivery_id))
|
||||
assert not queue.observe_client(ClientActivity(session_id="foreign", seq=2,
|
||||
recording=False, playback=False, muted=False, delivery_id=delivery_id))
|
||||
assert not queue.observe_client(activity(flow, 2, delivery_id="unknown"))
|
||||
assert not queue.observe_client(activity(flow, 3))
|
||||
assert queue.observe_client(activity(flow, 4, delivery_id=delivery_id))
|
||||
assert flow.state.delivery_floor.active_delivery_id == delivery_id
|
||||
assert queue.claim() is None # Idle observation is not delivery completion.
|
||||
|
||||
|
||||
def test_failure_or_close_never_changes_success_or_leaks_to_another_session():
|
||||
flow, turns, queue = setup()
|
||||
job = complete(flow)
|
||||
delivery_id = queue.enqueue(job)
|
||||
queue.claim()
|
||||
with pytest.raises(ValueError):
|
||||
queue.prepare(delivery_id, lambda _: "")
|
||||
assert flow.state.deliveries[delivery_id].status == "failed"
|
||||
second = complete(flow, "two")
|
||||
second_id = queue.enqueue(second)
|
||||
queue.claim()
|
||||
events = queue.prepare(second_id, lambda current: current[0].answer)
|
||||
third = complete(flow, "three")
|
||||
third_id = queue.enqueue(third)
|
||||
queue.close()
|
||||
assert not turns.accept_event(events[1])
|
||||
assert flow.state.deliveries[second_id].status == "interrupted"
|
||||
assert flow.state.deliveries[third_id].status == "skipped"
|
||||
assert queue.enqueue(job) is None
|
||||
assert queue.claim() is None
|
||||
other_flow, other_turns, other_queue = setup()
|
||||
assert other_queue.enqueue(second) is None
|
||||
assert not other_turns.accept_event(events[0])
|
||||
assert all(j.status == "completed" and j.answer for j in flow.state.jobs.values())
|
||||
|
||||
|
||||
def test_queue_requires_opt_in_composed_state():
|
||||
class Plain(Flow[TurnState]):
|
||||
conversational = True
|
||||
with pytest.raises(TypeError):
|
||||
ReplyQueue(TurnRunner(Plain()))
|
||||
|
||||
|
||||
def test_ended_conversation_cannot_enqueue_or_publish_results():
|
||||
flow, _, queue = setup()
|
||||
job = complete(flow)
|
||||
delivery_id = queue.enqueue(job)
|
||||
queue.claim()
|
||||
events = queue.prepare(delivery_id, lambda current: current[0].answer)
|
||||
assert events
|
||||
flow.state.ended = True
|
||||
assert not queue.accepts_output(delivery_id)
|
||||
assert queue.enqueue(job) is None
|
||||
assert queue.settle(delivery_id, "completed")
|
||||
assert flow.state.deliveries[delivery_id].status == "skipped"
|
||||
assert job.answer == "Result one"
|
||||
|
||||
|
||||
def test_shared_turn_interruption_fences_background_handoff_and_completion():
|
||||
flow, turns, queue = setup()
|
||||
job = complete(flow)
|
||||
delivery_id = queue.enqueue(job)
|
||||
record = queue.claim()
|
||||
events = queue.prepare(delivery_id, lambda current: current[0].answer)
|
||||
assert queue.accepts_output(delivery_id)
|
||||
assert turns.interrupt(session_id=record.session_id, turn_id=record.turn_id,
|
||||
input_revision=record.input_revision, delivery_id=record.delivery_id)
|
||||
assert not queue.accepts_output(delivery_id)
|
||||
assert not turns.accept_event(events[1])
|
||||
assert queue.settle(delivery_id, "completed")
|
||||
assert flow.state.deliveries[delivery_id].status == "interrupted"
|
||||
assert job.job_id not in flow.state.covered_updates
|
||||
assert job.status == "completed" and job.answer == "Result one"
|
||||
assert queue.claim() is None
|
||||
|
||||
|
||||
|
||||
def test_preparation_retry_during_playback_preserves_lease_and_completion():
|
||||
flow, turns, queue = setup()
|
||||
jobs = [complete(flow, name) for name in ("one", "two")]
|
||||
deliveries = [queue.enqueue(job) for job in jobs]
|
||||
delivery = queue.claim()
|
||||
events = queue.prepare(delivery.delivery_id, lambda current: current[0].answer)
|
||||
assert all(turns.accept_event(event) for event in events)
|
||||
assert queue.observe_client(ClientActivity(session_id=flow.state.id, seq=1,
|
||||
recording=False, playback=True, muted=False, delivery_id=delivery.delivery_id))
|
||||
snapshot = queue.snapshot()
|
||||
assert queue.prepare(delivery.delivery_id,
|
||||
lambda _: pytest.fail("Prepared twice")) == []
|
||||
assert queue.snapshot() == snapshot
|
||||
assert queue.accepts_output(delivery.delivery_id)
|
||||
assert queue.claim() is None
|
||||
assert queue.settle(delivery.delivery_id, "completed")
|
||||
assert flow.state.deliveries[delivery.delivery_id].status == "completed"
|
||||
assert not flow.state.replies[delivery.delivery_id].interrupted
|
||||
assert queue.claim() is None # Playback stop and completion are distinct.
|
||||
assert queue.observe_client(activity(flow, 2, delivery_id=delivery.delivery_id))
|
||||
assert queue.claim().delivery_id == deliveries[1]
|
||||
assert [job.status for job in jobs] == ["completed", "completed"]
|
||||
Reference in New Issue
Block a user