nat.builder.intermediate_step_manager

Attributes

logger

Classes

OpenStep
IntermediateStepManager	Manages updates to the NAT Event Stream for intermediate steps

Module Contents

logger

class OpenStep

step_id: str

step_name: str

step_type: str

step_parent_id: str

prev_stack: list[str]

active_stack: list[str]

class IntermediateStepManager(

context_state: nat.builder.context.ContextState,

)

Manages updates to the NAT Event Stream for intermediate steps

_instance_count: ClassVar[int] = 0

_active_instances: ClassVar[set[weakref.ref]]

_context_state

_outstanding_start_steps: dict[str, OpenStep]

push_intermediate_step(

payload: nat.data_models.intermediate_step.IntermediateStepPayload,

) → None

Pushes an intermediate step to the NAT Event Stream

push_intermediate_steps(

steps: list[nat.data_models.intermediate_step.IntermediateStep],

) → None

Inject a sequence of intermediate steps into the event stream without updating the step manager’s internal stack. Used to replay steps from a remote workflow (for example, from a /generate/full response) into the current workflow’s observability stream so the full tree is visible.

When replaying steps from a remote workflow, ensure the remote was invoked with the appropriate workflow-parent-id and workflow-parent-name (HTTP headers or session.run(parent_id=…, parent_name=…)) so the root step in the replayed list has the correct parent in the trace tree.

Parameters

stepslist[IntermediateStep]

Steps to inject (for example, parsed from a remote workflow response).

subscribe(

on_next: nat.utils.reactive.observable.OnNext[nat.data_models.intermediate_step.IntermediateStep],

on_error: nat.utils.reactive.observable.OnError = None,

on_complete: nat.utils.reactive.observable.OnComplete = None,

) → nat.utils.reactive.subscription.Subscription

Subscribes to the NAT Event Stream for intermediate steps

classmethod _cleanup_instance_tracking(ref: weakref.ref) → None

Cleanup callback for weakref when instance is garbage collected.

classmethod get_active_instance_count() → int

Get the number of active IntermediateStepManager instances.

Returns:

int: Number of active instances (cleaned up automatically via weakref)

get_outstanding_step_count() → int

Get the number of outstanding (started but not ended) steps.

Returns:

int: Number of steps that have been started but not yet ended