State stores¶
Persistence protocols and adapters for durable session state, stored events, and conversation bindings.
The ContextPatch object is documented under Routing policies.
Unified state store protocols and reference implementations.
Note: this module is intentionally lightweight to avoid import cycles.
The reference implementation lives in penguiflow.state.in_memory.
Bases: Protocol
Protocol for durable state adapters used by PenguiFlow.
Only the core audit-log methods are required. Additional subsystems detect
optional capabilities via duck-typing (hasattr / getattr).
async
¶
save_event(event: StoredEvent) -> None
Persist a runtime event.
Implementations may choose any storage backend (Postgres, Redis, etc.). The method must be idempotent since retries can emit duplicate events.
async
¶
load_history(trace_id: str) -> Sequence[StoredEvent]
Return the ordered history for a trace id.
async
¶
save_remote_binding(binding: RemoteBinding) -> None
Persist the mapping between a trace and an external worker.
dataclass
¶
StoredEvent(trace_id: str | None, ts: float, kind: str, node_name: str | None, node_id: str | None, payload: Mapping[str, Any])
Representation of a runtime event persisted by a state store.
Attributes:
| Name | Type | Description |
|---|---|---|
trace_id |
str | None
|
Trace id the event belongs to, if any. |
ts |
float
|
Unix timestamp (seconds) when the event occurred. |
kind |
str
|
Event type/kind, mirrors |
node_name |
str | None
|
Name of the node that emitted the event, if applicable. |
node_id |
str | None
|
Id of the node instance that emitted the event, if applicable. |
payload |
Mapping[str, Any]
|
Event-specific payload data. |
classmethod
¶
from_flow_event(event: FlowEvent) -> StoredEvent
Create a stored representation from a :class:~penguiflow.metrics.FlowEvent.
Bases: Protocol
Optional StateStore capability for tracking remote-agent conversation bindings.
A binding associates a router session with a remote agent/skill invocation so that follow-up turns can be routed back to the same remote task/context.
async
¶
find_binding(*, router_session_id: str, agent_url: str, remote_skill: str, tenant_id: str | None = None, user_id: str | None = None) -> RemoteBinding | None
Find an existing, non-terminal binding matching the given coordinates.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
router_session_id
|
str
|
Local session id issuing the remote call. |
required |
agent_url
|
str
|
URL of the remote agent. |
required |
remote_skill
|
str
|
Name of the remote skill/capability being invoked. |
required |
tenant_id
|
str | None
|
Optional tenant scope to match. |
None
|
user_id
|
str | None
|
Optional user scope to match. |
None
|
Returns:
| Type | Description |
|---|---|
RemoteBinding | None
|
The matching binding, or None if no such binding exists. |
async
¶
list_bindings(*, router_session_id: str) -> Sequence[RemoteBinding]
List all bindings recorded for a router session.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
router_session_id
|
str
|
Local session id to look up bindings for. |
required |
Returns:
| Type | Description |
|---|---|
Sequence[RemoteBinding]
|
The bindings associated with the session, in storage order. |
async
¶
mark_binding_terminal(*, trace_id: str, context_id: str | None, task_id: str) -> None
Mark a binding as terminal so it is no longer reused for follow-up turns.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
trace_id
|
str
|
Trace id of the binding to mark. |
required |
context_id
|
str | None
|
Optional remote context id associated with the binding. |
required |
task_id
|
str
|
Remote task id associated with the binding. |
required |
dataclass
¶
RemoteBinding(trace_id: str, context_id: str | None, task_id: str, agent_url: str, router_session_id: str | None = None, remote_skill: str | None = None, tenant_id: str | None = None, user_id: str | None = None, last_remote_task_id: str | None = None, is_terminal: bool = False, metadata: Mapping[str, Any] = dict())
Association between a trace and a remote worker/agent.
Attributes:
| Name | Type | Description |
|---|---|---|
trace_id |
str
|
Local trace id the binding is scoped to. |
context_id |
str | None
|
Remote context id, if the remote protocol uses one. |
task_id |
str
|
Remote task id assigned by the remote agent. |
agent_url |
str
|
URL of the remote agent handling the task. |
router_session_id |
str | None
|
Local session id that initiated the remote call, if any. |
remote_skill |
str | None
|
Name of the remote skill/capability invoked, if any. |
tenant_id |
str | None
|
Optional tenant scope for the binding. |
user_id |
str | None
|
Optional user scope for the binding. |
last_remote_task_id |
str | None
|
Most recent remote task id seen for follow-up turns, if any. |
is_terminal |
bool
|
Whether the binding has reached a terminal state and should no longer be reused. |
metadata |
Mapping[str, Any]
|
Additional free-form metadata about the binding. |
class-attribute
instance-attribute
¶
metadata: Mapping[str, Any] = field(default_factory=dict)
Persistence adapters for session/task state.
Bases: Protocol
async
¶
list_updates(session_id: str, *, task_id: str | None = None, since_id: str | None = None, limit: int = 500) -> Sequence[StateUpdate]
async
¶
list_steering(session_id: str, *, task_id: str | None = None, since_id: str | None = None, limit: int = 500) -> Sequence[SteeringEvent]
InMemorySessionStateStore(*, max_updates_per_session: int = 10000, max_steering_events_per_session: int = 10000)
Bases: SessionStateStore
In-memory state store for session/task persistence and replay.
async
¶
list_updates(session_id: str, *, task_id: str | None = None, since_id: str | None = None, limit: int = 500) -> Sequence[StateUpdate]
async
¶
list_steering(session_id: str, *, task_id: str | None = None, since_id: str | None = None, limit: int = 500) -> Sequence[SteeringEvent]
StateStoreSessionAdapter(store: StateStore)
Bases: SessionStateStore
Adapter that persists session updates into a core StateStore audit log.
async
¶
list_updates(session_id: str, *, task_id: str | None = None, since_id: str | None = None, limit: int = 500) -> Sequence[StateUpdate]
async
¶
list_steering(session_id: str, *, task_id: str | None = None, since_id: str | None = None, limit: int = 500) -> Sequence[SteeringEvent]
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.
Bases: BaseModel
An incremental update about a task's progress, persisted for later retrieval.
Used to stream status, tool-call, and result information for a task independent of the main event log, so clients can poll or subscribe to task-scoped updates.
class-attribute
instance-attribute
¶
session_id: str = Field(description='Session id the update belongs to.')
class-attribute
instance-attribute
¶
task_id: str = Field(description='Task id the update belongs to.')
class-attribute
instance-attribute
¶
trace_id: str | None = Field(default=None, description='Trace id associated with the update, if any.')
class-attribute
instance-attribute
¶
update_id: str = Field(default_factory=lambda: uuid.uuid4().hex, description='Unique identifier for this update.')
class-attribute
instance-attribute
¶
update_type: UpdateType = Field(description='Kind of update being recorded.')
class-attribute
instance-attribute
¶
content: Any = Field(description='Update payload; shape depends on `update_type`.')
class-attribute
instance-attribute
¶
step_index: int | None = Field(default=None, description='Zero-based index of this step, if the task reports progress by step.')
class-attribute
instance-attribute
¶
total_steps: int | None = Field(default=None, description='Total number of steps expected, if known.')
class-attribute
instance-attribute
¶
created_at: datetime = Field(default_factory=_utc_now, description='Timestamp the update was created.')
Bases: str, Enum
Kind of a StateUpdate emitted while a task runs.
Members
THINKING: Intermediate reasoning/plan content. PROGRESS: Free-form progress notification. TOOL_CALL: A tool invocation was made. RESULT: Final result content for a task. ERROR: An error occurred. CHECKPOINT: A resumable checkpoint was recorded. STATUS_CHANGE: The task's status changed. NOTIFICATION: A user-facing notification.