izWSSKJr SSKrSSKrSSKrSSKrSSKJrJrJ r J r J r J r J r SSKJrJr SSKJr SSKJr SSKJrJr SSKJrJrJr SS KJr SS KJr SS K J!r! SS K"J#r#J$r$ SS K%J&r& SSK'J(r( SSK)J*r*J+r+ SSK,J-r-J.r.J/r/ \(aSSK0J1r1 \ "S\S9r2\3\4\5\4r6"SS\75r8\Rr"S\85 "SS\ \-5r:\Rv"SS9SSj5r<\Rv"SS9SSj5r=g)) annotationsN) TYPE_CHECKINGAnyAsyncGeneratorGenericTypeTypeVarcast) BaseModelValidationError)SerializedContext) StepConfig)ContextSerdeErrorWorkflowRuntimeError)Event StartEvent StopEvent) BrokerState)WorkflowBroker) basic_runtime)PluginWorkflowRuntime) RunResultT)WorkflowHandler)BaseSerializerJsonSerializer)MODEL_T DictStateInMemoryStateStore)WorkflowT)boundc\rSrSrSrg)UnserializableKeyWarning5N)__name__ __module__ __qualname____firstlineno____static_attributes__r'S/home/james-whalen/.local/lib/python3.13/site-packages/workflows/context/context.pyr%r%5sr-r%oncec\rSrSr%SrSrS\S'S\S'S\S 'S \S 'S S \4SS jjr\ S Sj5r S!S"Sjjr S#S$Sjjr S%Sjr \ S&Sj5r\ S'Sj5rS!S(Sjjr\S!S)Sjj5rS*SjrS!S+SjjrS!S,SjjrS-S.SjjrS/SjrS0SjrS1Sjr\ S2Sj5rSrg )3Context<a 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 U=(d [5UlSUlX@lXlU=(d [5nUb/[ R "U5n[R"XQU5 O [ 5nXPl [5nUR5R5HtupU Rn U R cM U R [":wdM6[%U R [&5(dMWU R n UR)U 5 Mv [+U5S:a7[-SSR/UV s/sHoR0PM sn 5-5eU(aUR35O["n UR4(a}[6R8"UR4U5n U R:U :wa/[-SU R0SU R:R035e[=[6[>U 5Ul g[=[>U "55n [7U 5Ul g![an[SU35UeSnAff=fs sn f![Ban[ESU SU35UeSnAff=f) 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 _serializerr6r7r8r from_dict_autorfrom_serializedr r_init_snapshotset _get_stepsitems _step_configcontext_state_typer issubclassr addlen ValueErrorjoinr(popstater from_dict state_typer rr5 Exceptionr)selfworkflowprevious_context serializerpluginprevious_context_parsede state_types_ step_func step_configrK store_statestate_instances r.__init__Context.__init__sl&9)9 ! 3>#3  ' *;*J*J$+'+++z'8&9 #5 -0E $//1779LA&/&<&# TbTR5IShvN TbTR5IShvN ggN!![a N+f=fN7fN)shutdownrLacquire) prev_broker semaphoresr. before_start+Context._workflow_run..before_starts^&%..000$'')))%1 *s;A?=?AAA? A  A A  Ac6># TbTR5 gg7frk)release)rosr.after_complete-Context._workflow_run..after_completes$!!#%s)rNprevious start_eventrprtreturnNone)r6rgrr<r=r:start)rMrNrwrorprtrIrns ` @r. _workflow_runContext._workflow_runs7;    '**K#D ,,X6 * * $++   +;+; %%#%) &  r-c8URR5 g)z> Called internally from the handler to cancel a context's run N)_running_broker cancel_runr^s r._workflow_cancel_runContext._workflow_cancel_run s '')r-cJURc [S5eUR$)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.)r6rr^s r.rContext._running_brokers/    #&P r-cUR$)zTyped, 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. )r5r^s r.store Context.stores   r-clU=(d URn0nURbURRU5nURbURRnO,[ R "URURU5nURU5nX$l URSS9$)aSerialize 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) r:r5to_dictr6_staterr<r=r8 to_serializedrI model_dump)rMrP state_data broker_stateras r.rContext.to_dict$s< 34#3#3     (**22:>J    '++22L'66##T^^ZL,,Z8" !!x!00r-cRU"XUS9$![anSn[U5UeSnAff=f)aSReconstruct 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) ``` )rOrPzRError creating a Context instance: the provided payload has a wrong or old format.N)KeyErrorr)clsrNdatarPrSmsgs r.rJContext.from_dictXs7F 0x:N N 0fC#C(a / 0s &!&cR# URR5IShvN $N7f)zReturn the list of currently running step names. Returns: list[str]: Names of steps that have at least one active worker. N)r running_stepsr^s r.rContext.running_stepss" ))779999s '%'c:URRXU5$)a8 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] )rcollect_events)rMevexpected buffer_ids r.rContext.collect_eventssJ##222KKr-c8URRX5$)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 ``` )r send_event)rMmessagesteps r.rContext.send_eventsL##..w==r-cX# URRXX4U5IShvN $N7f)a0Wait 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)rwait_for_event)rM event_type waiter_event waiter_id requirementstimeouts r.rContext.wait_for_events2X))88 iw    s !*(*c:URRU5 g)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)rwrite_event_to_stream)rMrs r.rContext.write_event_to_streams 2226r-c[5 URRc [S5eURRR 5$)aReturn 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_resultr_handlerrresultr^s r. get_resultContext.get_resultsC(     ( ( 0&'DE E##,,3355r-c6URR5$)z8The internal queue used for streaming events to callers.)rstream_published_eventsr^s r. stream_eventsContext.stream_events2s##;;==r-c ^^[5 [R"5mSUU4Sjjn[R"U"55 T$![a/ [R "5nURU"55 T$f=f)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. c># TR5ShvN nTRU5IShvN [U[5(dM9 gN6N g7frk)rput isinstancer)rqrMs r._pump&Context.streaming_queue.._pump@sE ..0 beeBib),, 1s;AAAAAAA AAAArx)_warn_streaming_queueasyncioQueue create_task RuntimeErrorget_event_loop)rMrlooprs` @r.streaming_queueContext.streaming_queue6sp ")--/    &    ( &))+D   UW % &sA 5BB)r6r=r7r:r5r8) rNr!rOdict[str, Any] | NonerPBaseSerializer | NonerQrryrz)ryboolrk)rNr!rQzWorkflowRuntime | NoneryWorkflowBroker[MODEL_T])NN)rNr!rwzStartEvent | Nonerozasyncio.Semaphore | Noneryrrx)ryr)ryr4)rPrrydict[str, Any])rNz 'Workflow'rrrPrryz'Context[MODEL_T]')ryz list[str])rrrzlist[Type[Event]]r str | Noneryzlist[Event] | None)rrrrryrz)NNNi) rzType[T]r Event | Nonerrrrrz float | Noneryr")rrryrz)ryr)ryzAsyncGenerator[Event, None])ryz asyncio.Queue)r(r)r*r+__doc__known_unserializable_keys__annotations__rrZpropertyr]rgr|rrrr classmethodrJrrrrrrrrr,r'r-r.r1r1<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-r1)maxsizec8[R"S[SS9 g)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-c8[R"S[SS9 g)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-rx)> __future__rr functoolsrertypingrrrrrr r pydanticr r workflows.context.context_typesr workflows.decoratorsrworkflows.errorsrrworkflows.eventsrrr&workflows.runtime.types.internal_staterworkflows.runtime.brokerrworkflows.plugins.basicrworkflows.runtime.types.pluginrrworkflows.typesrworkflows.handlerr serializersrr state_storerrr workflowsr!r"dictrdlist EventBufferWarningr% simplefilterr1 lru_cacherrr'r-r.rs# 0=+ ?31B&-7??" Cu3U #$  w  f67OggOd Q    Q   r-