外观
typed_packets API
模块说明
text
Generic Core-backed packet codecs and ordered typed observations.
The Host owns wire encoding and connection context. No Python wire codec,
automatic retry, reconnect or background reader is installed by this module.MAX_FRAME_BYTES
python
MAX_FRAME_BYTES = 128 << 10MAX_DECODE_FRAME_BYTES
python
MAX_DECODE_FRAME_BYTES = 256 << 10EncodedPacket
python
@dataclass(frozen=True)
class EncodedPacketSDK 说明：
text
Diagnostic full plaintext frame and the actual context used by the Host.字段（包含继承字段）：
EncodedPacket.packet_id
python
packet_id: int数据类字段；声明于 typed_packets.EncodedPacket。
EncodedPacket.frame
python
frame: bytes数据类字段；声明于 typed_packets.EncodedPacket。
EncodedPacket.context
python
context: dict数据类字段；声明于 typed_packets.EncodedPacket。
TypedPacketObservation
python
class TypedPacketObservationSDK 说明：
text
One raw journal reader with typed models added to known packet events.
Every returned packet keeps its original raw frame and source metadata.
Unknown packet IDs remain raw. A failed known-packet decode seals the
stream; close remains available. Filtering never rewrites source cursors.TypedPacketObservation.__init__
python
def __init__(self, packets, raw_stream, snapshot, packet_ids=None)TypedPacketObservation.poll
python
async def poll(self)TypedPacketObservation.close
python
async def close(self)SDK 说明：
text
Join the in-flight read/decode; never cancel or silently reconnect.TypedPacketObservation.snapshot
python
self.snapshot公开实例属性，在 __init__ 中设置。
TypedPackets
python
class TypedPacketsSDK 说明：
text
Author facade for all registered schema packet models.
Codec success proves local encoding only. Send returns durable admission;
use operations.get/wait for dispatch evidence and preserve unknown results.TypedPackets.__init__
python
def __init__(self, context)TypedPackets.context_for
python
async def context_for(self, packet_type)SDK 说明：
text
Read the current connection codec context; grants remain Host-owned.TypedPackets.encode
python
async def encode(
self,
packet,
*,
context = 'offline',
connection_id = None,
offline_shield_id = None,
)SDK 说明：
text
Encode a diagnostic frame without admission or sending.
Offline encoding requires an explicit shield ID. Connection encoding
requires an epoch from context_for or observation; the Host owns all
runtime values. Never use diagnostic bytes as a typed send intent.TypedPackets.decode
python
async def decode(
self,
packet_type,
frame,
*,
context = 'offline',
connection_id = None,
offline_shield_id = None,
codec_revision = None,
)SDK 说明：
text
Decode a full plaintext frame into its generated packet model.
Captured connection frames require their codec_revision; never query
the current context to fill missing capture metadata.TypedPackets.decode_frame
python
async def decode_frame(self, frame, *, packet_id, connection_id, codec_revision)SDK 说明：
text
Decode one observed frame using its captured connection revision.TypedPackets.prepare
python
def prepare(self, packet, *, connection_id, idempotency_key, deadline)SDK 说明：
text
Freeze a model intent with zero IO; save it before awaiting send.TypedPackets.prepare_controlled
python
def prepare_controlled(
self,
packet,
*,
packet_type,
held_item_ref = None,
idempotency_key,
deadline,
)SDK 说明：
text
Freeze one of the nine permission-scoped action model intents.
Submit with ctx.operations.submit/execute. The Host checks the closed
action profile, identity and current inventory before writing. Item
fields must stay at their default zero placeholder; held_item_ref binds
the authoritative native slot. This method performs no codec RPC or IO.TypedPackets.send
python
async def send(self, intent, *, timeout=10)SDK 说明：
text
Submit once; return an operation ID for explicit reconciliation.TypedPackets.subscribe
python
def subscribe(
self,
packet_types,
*,
max_events = 64,
max_bytes = 524288,
poll_interval = 0.025,
)SDK 说明：
text
Select inbound generated models with per-ID observation grants.
Use ``async with ... as stream`` and ``async for event in stream``.
Requires Host API 1.5; does not fall back to raw diagnostic observation.TypedPackets.open
python
async def open(self, packet_types=None, *, direction='inbound')SDK 说明：
text
Open a connection-bound stream, preserving raw unknown packet IDs.
Optional packet_types selects generated Packet subclasses. Requires
the Host's raw direction grant plus its generic decode permissions.TypedPackets.listen
python
async def listen(self, packet_types, callback, *, direction='inbound', on_snapshot=None)SDK 说明：
text
Run serial callbacks in a Worker-owned task; cancel and await to close.TypedPackets.context
python
self.context公开实例属性，在 __init__ 中设置。