JiWddlmZddlZddlZddlZddlZddlmZmZm Z m Z m Z m Z m Z ddlmZmZddlmZddlmZddlmZmZddlmZmZmZdd lmZdd lmZdd l m!Z!dd l"m#Z#dd l$m%Z%m&Z&ddl'm(Z(ddl)m*Z*m+Z+ddl,m-Z-m.Z.m/Z/erddl0m1Z1e deZ2e3e4e5efZ6Gdde7Z8ejrde8Gdde e-Z:ejvdddZ<ejvdddZ=y)) annotationsN) TYPE_CHECKINGAnyAsyncGeneratorGenericTypeTypeVarcast) BaseModelValidationError)SerializedContext) StepConfig)ContextSerdeErrorWorkflowRuntimeError)Event StartEvent StopEvent)WorkflowHandler) basic_runtime)WorkflowBroker) BrokerState)PluginWorkflowRuntime) RunResultT)BaseSerializerJsonSerializer)MODEL_T DictStateInMemoryStateStore)WorkflowT)boundc eZdZy)UnserializableKeyWarningN)__name__ __module__ __qualname__k/var/www/html/BankruptcyAI-uat/bankruptcy-ai/venv/lib/python3.12/site-packages/workflows/context/context.pyr%r%4sr*r%oncec|eZdZUdZdZded<ded<ded<d ed <d d ef dd Zedd Z d d dZ d! d"dZ d#dZ ed$dZ ed%dZdd&dZe d d'dZd(dZ d d)dZdd*dZ d+ d,dZd-dZd.dZd/dZed0dZy )1Contexta Global, per-run context for a `Workflow`. Provides an interface into the underlying broker run, for both external (workflow run oberservers) and internal consumption by workflow steps. The `Context` coordinates event delivery between steps, tracks in-flight work, exposes a global state store, and provides utilities for streaming and synchronization. It is created by a `Workflow` at run time and can be persisted and restored. Args: workflow (Workflow): The owning workflow instance. Used to infer step configuration and instrumentation. previous_context: A previous context snapshot to resume from. serializer: A serializer to use for serializing and deserializing the current and previous context snapshots. Attributes: is_running (bool): Whether the workflow is currently running. store (InMemoryStateStore[MODEL_T]): Type-safe, async state store shared across steps. See also [InMemoryStateStore][workflows.context.state_store.InMemoryStateStore]. Examples: Basic usage inside a step: ```python from workflows import step from workflows.events import StartEvent, StopEvent @step async def start(self, ctx: Context, ev: StartEvent) -> StopEvent: await ctx.store.set("query", ev.topic) ctx.write_event_to_stream(ev) # surface progress to UI return StopEvent(result="ok") ``` Persisting the state of a workflow across runs: ```python from workflows import Context # Create a context and run the workflow with the same context ctx = Context(my_workflow) result_1 = await my_workflow.run(..., ctx=ctx) result_2 = await my_workflow.run(..., ctx=ctx) # Serialize the context and restore it ctx_dict = ctx.to_dict() restored_ctx = Context.from_dict(my_workflow, ctx_dict) result_3 = await my_workflow.run(..., ctx=restored_ctx) ``` See Also: - [Workflow][workflows.Workflow] - [Event][workflows.events.Event] - [InMemoryStateStore][workflows.context.state_store.InMemoryStateStore] )memoryInMemoryStateStore[MODEL_T] _state_storezWorkflowBroker[MODEL_T] | None _broker_runr_pluginr! _workflowNc |xs t|_d|_||_||_|xs t}|. t j |}tj|||n t }||_ t}|jjD]j\}} | j} | j | j t"k7s3t%| j t&sN| j } |j)| lt+|dkDr5t-ddj/|D cgc]} | j0c} z|r|j3nt"} |j4r{t7j8|j4|} | j:| k7r/t-d| j0d| j:j0t=t6t>| |_ y t=t>| } t7| |_ y#t$r}td||d}~wwxYwcc} w#tB$r}tEd| d||d}~wwxYw) Nz-Context dict specified in an invalid format: rzqMultiple state types are not supported. Make sure that each Context[...] has the same generic state type. Found: z, z/State type mismatch. Workflow context expected z, got z#Failed to initialize state of type zB. Does your state define defaults for all fields? Original error: )#r _serializerr2r3r4r from_dict_autorfrom_serializedr r_init_snapshotset _get_stepsitems _step_configcontext_state_typer issubclassr addlen ValueErrorjoinr&popstater from_dict state_typer rr1 Exceptionr)selfworkflowprevious_context serializerpluginprevious_context_parsede state_types_ step_func step_configrG store_statestate_instances r+__init__zContext.__init__sv&9)9 ! 3>#3  ' *;*J*J$+'+++Xz'8&9 #5 -0E $//1779 ,LAy&/&<&.before_starts_&%..000$'')))%1 *s;A=;=AA A= A AA  Ac4Kjyywre)release)risr+after_completez-Context._workflow_run..after_completes$!!#%s)rJprevious start_eventrjrmreturnNone)r2rbrr8r9r6start)rIrJrorirjrmrErhs ` @r+ _workflow_runzContext._workflow_runs7;    '**K#D ,,X6 * $++   4+;+; %%#%) &  r*c8|jjy)zN Called internally from the handler to cancel a context's run N)_running_broker cancel_runrYs r+_workflow_cancel_runzContext._workflow_cancel_run s '')r*cH|j td|jS)Nz}Workflow run is not yet running. Make sure to only call this method after the context has been passed to a workflow.run call.)r2rrYs r+rvzContext._running_brokers/    #&P r*c|jS)aTyped, process-local state store shared across steps. If no state was initialized yet, a default [DictState][workflows.context.state_store.DictState] store is created. Returns: InMemoryStateStore[MODEL_T]: The state store instance. )r1rYs r+storez Context.stores   r*cd|xs |j}i}|j|jj|}|j|jj}n+t j |j|j|}|j|}||_ |jdS)aiSerialize the context to a JSON-serializable dict. Persists the global state store, event queues, buffers, accepted events, broker log, and running flag. This payload can be fed to [from_dict][workflows.context.context.Context.from_dict] to resume a run or carry state across runs. Args: serializer (BaseSerializer | None): Value serializer used for state and event payloads. Defaults to [JsonSerializer][workflows.context.serializers.JsonSerializer]. Returns: dict[str, Any]: A dict suitable for JSON encoding and later restoration via `from_dict`. See Also: - [InMemoryStateStore.to_dict][workflows.context.state_store.InMemoryStateStore.to_dict] Examples: ```python ctx_dict = ctx.to_dict() my_db.set("key", json.dumps(ctx_dict)) ctx_dict = my_db.get("key") restored_ctx = Context.from_dict(my_workflow, json.loads(ctx_dict)) result = await my_workflow.run(..., ctx=restored_ctx) ``` python)mode) r6r1to_dictr2_staterr8r9r4 to_serializedrE model_dump)rIrL state_data broker_stater[s r+rzContext.to_dict$s< 34#3#3     (**22:>J    '++22L'66##T^^ZL,,Z8" !!x!00r*cX ||||S#t$r}d}t||d}~wwxYw)aReconstruct a `Context` from a serialized payload. Args: workflow (Workflow): The workflow instance that will own this context. data (dict[str, Any]): Payload produced by [to_dict][workflows.context.context.Context.to_dict]. serializer (BaseSerializer | None): Serializer used to decode state and events. Defaults to JSON. Returns: Context[MODEL_T]: A context instance initialized with the persisted state and queues. Raises: ContextSerdeError: If the payload is missing required fields or is in an incompatible format. Examples: ```python ctx_dict = ctx.to_dict() my_db.set("key", json.dumps(ctx_dict)) ctx_dict = my_db.get("key") restored_ctx = Context.from_dict(my_workflow, json.loads(ctx_dict)) result = await my_workflow.run(..., ctx=restored_ctx) ``` )rKrLzRError creating a Context instance: the provided payload has a wrong or old format.N)KeyErrorr)clsrJdatarLrOmsgs r+rFzContext.from_dictXs9F 0x$:N N 0fC#C(a / 0s )$)cRK|jjd{S7w)zReturn the list of currently running step names. Returns: list[str]: Names of steps that have at least one active worker. N)rv running_stepsrYs r+rzContext.running_stepss# ))779999s '%'c<|jj|||S)a Buffer events until all expected types are available, then return them. This utility is helpful when a step can receive multiple event types and needs to proceed only when it has a full set. The returned list is ordered according to `expected`. Args: ev (Event): The incoming event to add to the buffer. expected (list[Type[Event]]): Event types to collect, in order. buffer_id (str | None): Optional stable key to isolate buffers across steps or workers. Defaults to an internal key derived from the task name or expected types. Returns: list[Event] | None: The events in the requested order when complete, otherwise `None`. Examples: ```python @step async def synthesize( self, ctx: Context, ev: QueryEvent | RetrieveEvent ) -> StopEvent | None: events = ctx.collect_events(ev, [QueryEvent, RetrieveEvent]) if events is None: return None query_ev, retrieve_ev = events # ... proceed with both inputs present ... ``` See Also: - [Event][workflows.events.Event] )rvcollect_events)rIevexpected buffer_ids r+rzContext.collect_eventss J##222xKKr*c:|jj||S)aGDispatch an event to one or all workflow steps. If `step` is omitted, the event is broadcast to all step queues and non-matching steps will ignore it. When `step` is provided, the target step must accept the event type or a [WorkflowRuntimeError][workflows.errors.WorkflowRuntimeError] is raised. Args: message (Event): The event to enqueue. step (str | None): Optional step name to target. Raises: WorkflowRuntimeError: If the target step does not exist or does not accept the event type. Examples: It's common to use this method to fan-out events: ```python @step async def my_step(self, ctx: Context, ev: StartEvent) -> WorkerEvent | GatherEvent: for i in range(10): ctx.send_event(WorkerEvent(msg=i)) return GatherEvent() ``` You also see this method used from the caller side to send events into the workflow: ```python handler = my_workflow.run(...) async for ev in handler.stream_events(): if isinstance(ev, SomeEvent): handler.ctx.send_event(SomeOtherEvent(msg="Hello!")) result = await handler ``` )rv send_event)rImessagesteps r+rzContext.send_eventsL##..w==r*c\K|jj|||||d{S7w)a Wait for the next matching event of type `event_type`. Optionally emits a `waiter_event` to the event stream once per `waiter_id` to inform callers that the workflow is waiting for external input. This helps to prevent duplicate waiter events from being sent to the event stream. Args: event_type (type[T]): Concrete event class to wait for. waiter_event (Event | None): Optional event to write to the stream once when the wait begins. waiter_id (str | None): Stable identifier to avoid emitting multiple waiter events for the same logical wait. requirements (dict[str, Any] | None): Key/value filters that must be satisfied by the event via `event.get(key) == value`. timeout (float | None): Max seconds to wait. `None` means no timeout. Defaults to 2000 seconds. Returns: T: The received event instance of the requested type. Raises: asyncio.TimeoutError: If the timeout elapses. Examples: ```python @step async def my_step(self, ctx: Context, ev: StartEvent) -> StopEvent: response = await ctx.wait_for_event( HumanResponseEvent, waiter_event=InputRequiredEvent(msg="What's your name?"), waiter_id="user_name", timeout=60, ) return StopEvent(result=response.response) ``` N)rvwait_for_event)rI event_type waiter_event waiter_id requirementstimeouts r+rzContext.wait_for_events7X))88  iw    s #,*,c:|jj|y)aEnqueue an event for streaming to [WorkflowHandler]](workflows.handler.WorkflowHandler). Args: ev (Event | None): The event to stream. `None` can be used as a sentinel in some streaming modes. Examples: ```python @step async def my_step(self, ctx: Context, ev: StartEvent) -> StopEvent: ctx.write_event_to_stream(ev) return StopEvent(result="ok") ``` N)rvwrite_event_to_stream)rIrs r+rzContext.write_event_to_streams 2226r*ct|jj td|jjj S)a<Return the final result of the workflow run. Deprecated: This method is deprecated and will be removed in a future release. Prefer awaiting the handler returned by `Workflow.run`, e.g.: `result = await workflow.run(..., ctx=ctx)`. Examples: ```python # Preferred result = await my_workflow.run(..., ctx=ctx) # Deprecated result_agent = ctx.get_result() ``` Returns: RunResultT: The value provided via a `StopEvent`. zWorkflow handler is not set)_warn_get_resultrv_handlerrresultrYs r+ get_resultzContext.get_resultsC(     ( ( 0&'DE E##,,3355r*c6|jjS)z8The internal queue used for streaming events to callers.)rvstream_published_eventsrYs r+ stream_eventszContext.stream_events2s##;;==r*cttjdfd } tj|S#t$r.tj }|j|YSwxYw)zDeprecated queue-based event stream. Returns an asyncio.Queue that is populated by iterating this context's stream_events(). A deprecation warning is emitted once per process. cKj23d{}j|d{t|ts3y7076ywre)rput isinstancer)rqrIs r+_pumpz&Context.streaming_queue.._pump@sJ ..0  beeBib), 1s;AA A A AA AA A  A Arp)_warn_streaming_queueasyncioQueue create_task RuntimeErrorget_event_loop)rIrlooprs` @r+streaming_queuezContext.streaming_queue6sk ")--/   &    ( &))+D   UW % &sA3A<;A<) rJr!rKdict[str, Any] | NonerLBaseSerializer | NonerMrrqrr)rqboolre)rJr!rMzWorkflowRuntime | NonerqWorkflowBroker[MODEL_T])NN)rJr!rozStartEvent | Nonerizasyncio.Semaphore | Nonerqrrp)rqr)rqr0)rLrrqdict[str, Any])rJz 'Workflow'rrrLrrqz'Context[MODEL_T]')rqz list[str])rrrzlist[Type[Event]]r str | Nonerqzlist[Event] | None)rrrrrqrr)NNNi) rzType[T]r Event | Nonerrrrrz float | Nonerqr")rrrqrr)rqr)rqzAsyncGenerator[Event, None])rqz asyncio.Queue)r&r'r(__doc__known_unserializable_keys__annotations__rrVpropertyrXrbrtrxrvr{r classmethodrFrrrrrrrrr)r*r+r.r.;s9z!,.-// O 37,0& FF0F* F  F  FP//DH *@ (*..2 & & '& , &  & P*    ! !21h -1 &0&0&0* &0  &0&0P:OS%L%L#4%LAK%L %LN&>V&* $.2 $ . . #.  . , .  .  . `7"62>r*r.)maxsizec<tjdtdy)NzContext.get_result() is deprecated and will be removed in a future release. Prefer awaiting the WorkflowHandler returned by Workflow.run: `result = await workflow.run(..., ctx=ctx)`. stacklevelwarningswarnDeprecationWarningr)r*r+rrNs MM I r*c<tjdtdy)NzContext.streaming_queue is deprecated and will be removed in a future release. Prefer iterating Context.stream_events(): `async for ev in ctx.stream_events(): ...`rrrr)r*r+rr[s MM 9 r*rp)> __future__rr functoolsr_rtypingrrrrrr r pydanticr r workflows.context.context_typesr workflows.decoratorsrworkflows.errorsrrworkflows.eventsrrrworkflows.handlerrworkflows.plugins.basicrworkflows.runtime.brokerr&workflows.runtime.types.internal_staterworkflows.runtime.types.pluginrrworkflows.typesr serializersrr state_storerrr workflowsr!r"dictr^list EventBufferWarningr% simplefilterr. lru_cacherrr)r*r+rs# 0=+ .13>B&7??" Cu3U #$  w f67PggPfQ   Q   r*