Routing policies

Config-driven routing: the policy protocol, its dictionary-backed implementation, the routing request, and the context-patch objects used to mutate routing state.

policies

Policy helpers for dynamic routing decisions.

RoutingPolicy

Bases: Protocol

Protocol for routing policies used by router nodes.

Implementations decide, for a given RoutingRequest, which downstream node(s) a message should be routed to.

select

select(request: RoutingRequest) -> RoutingDecisionType | Awaitable[RoutingDecisionType]

Return the desired routing targets for request.

Parameters:

Name	Type	Description	Default
request	RoutingRequest	The routing request describing the message, node, and proposed candidates.	required

Returns:

Type	Description
RoutingDecisionType | Awaitable[RoutingDecisionType]	None to accept the proposed routing, a single Node or node name, a
RoutingDecisionType | Awaitable[RoutingDecisionType]	sequence of nodes/names to route to, or an awaitable resolving to one of
RoutingDecisionType | Awaitable[RoutingDecisionType]	these.

DictRoutingPolicy

DictRoutingPolicy(mapping: Mapping[str, RoutingDecisionType], *, default: RoutingDecisionType = None, key_getter: KeyFn | None = None)

Routing policy driven by a mapping loaded from config.

Selection uses key_getter (the trace id by default) to look up a routing decision in the configured mapping, falling back to default when the key is absent or unmapped.

Initialize the policy with a mapping and optional default/key function.

Parameters:

Name	Type	Description	Default
mapping	Mapping[str, RoutingDecisionType]	Mapping from lookup key to routing decision.	required
default	RoutingDecisionType	Decision to return when the key is missing or unmapped.	None
key_getter	KeyFn | None	Callable deriving the lookup key from a RoutingRequest; defaults to using request.trace_id.	None

select

select(request: RoutingRequest) -> RoutingDecisionType

Return the routing decision configured for request's key.

Parameters:

Name	Type	Description	Default
request	RoutingRequest	The routing request to evaluate.	required

Returns:

Type	Description
RoutingDecisionType	The mapped routing decision, or the configured default if the key is
RoutingDecisionType	None or not present in the mapping.

update_mapping

update_mapping(mapping: Mapping[str, RoutingDecisionType]) -> None

Replace the policy's mapping in place.

Parameters:

Name	Type	Description	Default
mapping	Mapping[str, RoutingDecisionType]	The new mapping from lookup key to routing decision.	required

set_default

set_default(decision: RoutingDecisionType) -> None

Set the fallback decision used when a key is missing or unmapped.

Parameters:

Name	Type	Description	Default
decision	RoutingDecisionType	The new default routing decision.	required

from_json classmethod

from_json(payload: str, **kwargs: Any) -> DictRoutingPolicy

Build a policy from a JSON string that decodes to a mapping.

Parameters:

Name	Type	Description	Default
payload	str	JSON-encoded mapping of lookup key to routing decision.	required
**kwargs	Any	Additional keyword arguments forwarded to the constructor (e.g. default, key_getter).	{}

Returns:

Type	Description
DictRoutingPolicy	A new DictRoutingPolicy built from the decoded mapping.

Raises:

Type	Description
TypeError	If the decoded JSON payload is not a mapping.

from_json_file classmethod

from_json_file(path: str, **kwargs: Any) -> DictRoutingPolicy

Build a policy from a JSON file that decodes to a mapping.

Parameters:

Name	Type	Description	Default
path	str	Path to a UTF-8 encoded JSON file.	required
**kwargs	Any	Additional keyword arguments forwarded to the constructor (e.g. default, key_getter).	{}

Returns:

Type	Description
DictRoutingPolicy	A new DictRoutingPolicy built from the file's decoded mapping.

Raises:

Type	Description
TypeError	If the decoded JSON payload is not a mapping.

from_env classmethod

from_env(env_var: str, *, loader: Callable[[str], Mapping[str, RoutingDecisionType]] | None = None, default: RoutingDecisionType = None, key_getter: KeyFn | None = None) -> DictRoutingPolicy

Build a policy from an environment variable.

Parameters:

Name	Type	Description	Default
env_var	str	Name of the environment variable holding the policy data.	required
loader	Callable[[str], Mapping[str, RoutingDecisionType]] | None	Optional callable to parse the raw environment value into a mapping; defaults to json.loads.	None
default	RoutingDecisionType	Decision to return when the key is missing or unmapped.	None
key_getter	KeyFn | None	Callable deriving the lookup key from a RoutingRequest; defaults to using request.trace_id.	None

Returns:

Type	Description
DictRoutingPolicy	A new DictRoutingPolicy built from the parsed mapping.

Raises:

Type	Description
KeyError	If the environment variable is not set.
TypeError	If the parsed data is not a mapping.

RoutingRequest dataclass

RoutingRequest(message: Any, context: Context, node: Node, proposed: tuple[Node, ...], trace_id: str | None)

Information provided to routing policies.

Attributes:

Name	Type	Description
message	Any	The message payload being routed.
context	Context	The runtime Context in which routing is occurring.
node	Node	The router node making the routing decision.
proposed	tuple[Node, ...]	The candidate downstream nodes proposed by the router.
trace_id	str | None	Identifier of the trace being routed, if known.

message instance-attribute

message: Any

context instance-attribute

context: Context

node instance-attribute

node: Node

proposed instance-attribute

proposed: tuple[Node, ...]

trace_id instance-attribute

trace_id: str | None

node_name property

node_name: str

Return the router node's name, falling back to its node id.

proposed_names property

proposed_names: tuple[str, ...]

Return the display names of the proposed candidate nodes.

Returns:

Type	Description
str	A tuple of names (or node ids when a candidate has no name), in the same
...	order as proposed.

models

Session/task models for bidirectional streaming and background work.

Most persistence-facing task/steering models live in penguiflow.state.models. This module re-exports them for backward compatibility.

ContextPatch

Bases: BaseModel

Structured summary of a completed background task's contribution to context.

Produced when a background task finishes and needs to hand its results back to the foreground agent's context, either immediately (APPEND/REPLACE) or pending human approval (HUMAN_GATED).

task_id class-attribute instance-attribute

task_id: str = Field(description='Identifier of the task that produced this patch.')

spawned_from_event_id class-attribute instance-attribute

spawned_from_event_id: str | None = Field(default=None, description='Event ID of the foreground turn that spawned the originating task, if any.')

source_context_version class-attribute instance-attribute

source_context_version: int | None = Field(default=None, description='Context version the task branched from, for divergence detection.')

source_context_hash class-attribute instance-attribute

source_context_hash: str | None = Field(default=None, description='Hash of the context the task branched from, for divergence detection.')

context_diverged class-attribute instance-attribute

context_diverged: bool = Field(default=False, description='Whether the foreground context changed since the task was spawned.')

completed_at class-attribute instance-attribute

completed_at: datetime = Field(default_factory=_utc_now, description='Timestamp when the task completed.')

digest class-attribute instance-attribute

digest: list[str] = Field(default_factory=list, description='Short human-readable summary lines of the result.')

facts class-attribute instance-attribute

facts: dict[str, Any] = Field(default_factory=dict, description='Structured facts extracted from the task run.')

artifacts class-attribute instance-attribute

artifacts: list[dict[str, Any]] = Field(default_factory=list, description='Artifacts (files, links, structured payloads) produced by the task.')

sources class-attribute instance-attribute

sources: list[dict[str, Any]] = Field(default_factory=list, description='Source references (documents, URLs) consulted during the task.')

recommended_next_steps class-attribute instance-attribute

recommended_next_steps: list[str] = Field(default_factory=list, description='Suggested follow-up actions surfaced by the task.')

assumptions class-attribute instance-attribute

assumptions: list[str] = Field(default_factory=list, description='Assumptions the task made while producing its result.')

session

Streaming session orchestration for bidirectional tasks.

PendingContextPatch dataclass

PendingContextPatch(patch_id: str, task_id: str, patch: ContextPatch, strategy: MergeStrategy, created_at: float = time.time())

A context patch awaiting human approval (MergeStrategy.HUMAN_GATED).

Attributes:

Name	Type	Description
patch_id	str	Unique identifier used to approve or reject this patch via steer/apply_pending_patch.
task_id	str	The task whose completion produced this patch.
patch	ContextPatch	The underlying :class:ContextPatch payload to merge if approved.
strategy	MergeStrategy	The merge strategy the patch was submitted with.
created_at	float	Unix timestamp (seconds) when the patch was queued.

patch_id instance-attribute

patch_id: str

task_id instance-attribute

task_id: str

patch instance-attribute

patch: ContextPatch

strategy instance-attribute

strategy: MergeStrategy

created_at class-attribute instance-attribute

created_at: float = field(default_factory=time.time)