band-sdk-core

Shared Band event-payload validation, memory-taxonomy, and inbound-delivery runtime state, compiled from Rust into a native Python extension (import band_sdk_core). Every function is synchronous, pure computation — no network, filesystem, or logging.

band-sdk-core-core (Rust)  -->  band-sdk-core (this package, PyPI)  -->  band-sdk-python

If you're using band-sdk-python, it already depends on this package — you don't need to install it yourself. Install it directly only if you're calling into it without that SDK. Full picture: repository architecture.

Install

pip install band-sdk-core

Wheels are prebuilt (abi3, Python >= 3.11) for Linux (manylinux/musllinux, x86_64/aarch64), macOS (arm64/x86_64), and Windows (amd64/arm64) — no Rust toolchain needed to install.

Quickstart

import band_sdk_core

try:
    payload = band_sdk_core.validate_event_payload("room_deleted", {"id": "room-1"})
except ValueError as e:
    for path, code, message in e.issues:
        print(f"{path}: {code} - {message}")

validate_event_payload is the platform's inbound WebSocket payload policy: normalize a well-formed payload, or reject a malformed one with every violation reported at once, never just the first.

Public surface

validate_event_payload(event_type, raw, trace_context=None)

event_type is a platform name ("message_created", "agent.control", …) or an EventType. raw is a JSON-shaped Python value. Success returns the normalized payload (event_created returns the input unchanged). Failure raises ValueError with .args == (message,) (str(e) is ordinary), .issues (a tuple of (path, code, message)), and .trace_context. There is no custom exception class.

EventType

The closed set of inbound event names.

Delivery-state runtime classes

ClaimRegistry, RetryTracker, ParticipantRoster, and SubscriptionTracker are the inbound-delivery runtime state classes — lifecycle and design decisions live as rustdoc on each type in crates/core/src/runtime/.

A participant is any mapping; add/set_all read id, name, type, handle, description from it and list() returns dicts with exactly those five keys. id is required and must be a string; the other four are each a string or None. A non-mapping value, a missing/non-string id, or another field that is not a string or None raises TypeError. set_all takes an optional trace_context and raises ValueError — leaving the roster unchanged, with .issues and .trace_context attached like validate_event_payload's error — if its snapshot names the same id twice. RetryTracker's max_tracked must be at least 1; 0 raises ValueError. ClaimRegistry's max_completed must be at least 1 too — 0 also raises ValueError.

ClaimRegistry keeps one terminal intent per message until remember_completed or matching confirm_settlement: remember_ack_pending records a processed write and remember_failed_ack_pending a failed one carrying its error. The first intent wins; a different one raises ValueError — leaving the entry unchanged, with .issues and .trace_context attached — and an identical repeat is a no-op. pending_ack returns {"status", "error"?, "parked"} or None. park_ack takes a ParkedReason and raises TypeError for any other value, including its wire-name string; ParkedReason.from_wire_name converts.

SubscriptionTracker provides synchronous, transport-independent decisions for agent-topic joins and the two-topic room subscription transaction. It returns opaque integer tickets; callers supply the matching ticket when recording each completion. Failed rollbacks and failed or unknown leaves require explicit reconciliation before a fresh claim is allowed.

SubscriptionTracker lifecycle

from band_sdk_core import LeaveOutcome, RoomStatus, RoomSubscribeResult, SubscriptionTracker

tracker = SubscriptionTracker()
ticket = tracker.begin_room_subscribe("room-1")
if ticket is None:
    raise RuntimeError("room is not claimable")

result = tracker.record_room_participants_join_failed("room-1", ticket, False)
if result is RoomSubscribeResult.RollbackFailed:
    assert tracker.room_status("room-1") is RoomStatus.NeedsReconciliation
    assert tracker.acknowledge_room_reconciled("room-1") is True
    ticket = tracker.begin_room_subscribe("room-1")
    assert ticket is not None
    assert tracker.record_both_room_topics_joined("room-1", ticket) is RoomSubscribeResult.Subscribed
    leave_ticket = tracker.unsubscribe_room("room-1")
    assert leave_ticket is not None
    assert tracker.mark_room_leave_complete("room-1", leave_ticket, LeaveOutcome.Left) is True
else:
    match result:
        case (
            RoomSubscribeResult.Subscribed
            | RoomSubscribeResult.JoinFailed
            | RoomSubscribeResult.RolledBack
            | RoomSubscribeResult.Stale
        ):
            pass
        case _:
            raise AssertionError(f"unhandled subscription result: {result}")

DecisionRegistry lifecycle

DecisionRegistry tracks chat-mediated asks: whoever claims one first owns its outcome. Every registration returns (token, ticket), and later calls name that exact registration, so a stale holder gets ClaimOutcome.Stale instead of acting on a newer ask that reused the token. max_pending must be at least 1, and seed a non-negative 64-bit integer; either out of range raises ValueError. An omitted seed draws a fresh one. is_authorized_sender takes a set, frozenset, or None.

from band_sdk_core import ClaimOutcome, DecisionRegistry, is_authorized_sender

registry = DecisionRegistry(max_pending=10)
registration = registry.register_minted("room-1")
assert registration is not None
token, ticket = registration

assert registry.unclaimed_in_room("room-1") == [token]
assert registry.try_claim(token, ticket) is ClaimOutcome.Claimed
assert registry.try_claim(token, ticket) is ClaimOutcome.AlreadyClaimed
assert registry.cancel_room("room-1").claimed == [token]
assert is_authorized_sender(frozenset({"alice"}), "alice") is True

Session — WebSocket reconnect state machine

Session/SessionPolicy are a sans-io session state machine plus reconnect backoff/jitter policy; classify_close/classify_upgrade classify a WebSocket close code or HTTP upgrade-rejection status. Session never sleeps, connects, or closes a socket itself — the caller drives its own transport and reports what happened through on_connected/ on_socket_close/on_upgrade_rejected/on_supersede. Decisions live as rustdoc in crates/core/src/runtime/session.rs.

Session lifecycle

from band_sdk_core import Session, SessionPolicy, SessionState

session = Session(SessionPolicy.default())
epoch = session.begin_attempt(0.0)
assert epoch is not None

connected = session.on_connected(epoch, 0.0)
assert connected.state is SessionState.Up

disconnected = session.on_socket_close(epoch, 5.0, 1006, 0.5)
assert disconnected.state is SessionState.Reconnecting
assert disconnected.retry_after_s is not None

Memory taxonomy

MemorySystem, MemoryType, MemorySegment, MemoryStoreScope, MemoryListScope, MemoryStatus are the canonical memory taxonomy — design decisions live as rustdoc in crates/core/src/memory.rs. Each is a closed set with a wire_name property and a from_wire_name static method (None for an unrecognized string), mirroring EventType.

validate_memory_type_for_system(system, memory_type, trace_context=None)

system/memory_type are each a wire-name string or the matching MemorySystem/MemoryType instance (mirroring validate_event_payload's acceptance of either an event name or an EventType), and it returns None on success. Failure raises ValueError with the same .issues/.trace_context contract as validate_event_payload — an unrecognized system and an unrecognized memory_type are independent issues, both reported when both strings are invalid.

One-shot delivery lifecycle

evaluate_delivery_event, evaluate_next_message, evaluate_drain_candidate, and evaluate_adapter_result are the stateless one-shot delivery lifecycle decisions — design decisions live as rustdoc in crates/core/src/runtime/delivery.rs. Each returns a plain dict tagged by its own "decision" string; there is no cross-invocation state, so there is no class here to construct.

evaluate_delivery_event(event_type, room_id, payload, agent_id, trace_context=None) routes one inbound event: "ignored" for an unrecognized or out-of-scope event_type; "cleanup" for room_removed/room_deleted; "skip_self" or "invocation" for message_created, depending on whether the sender is this agent (sender_type == "Agent" and sender_id == agent_id). message_created delegates to validate_event_payload for payload-shape validation, so a malformed payload raises the same ValueError, and additionally rejects an empty payload["id"] — a message with no identity cannot be claimed or acknowledged. room_removed/room_deleted need no payload shape and read id straight off the raw payload.

room_id resolves from the caller-supplied room_id first, else the payload's own chat_room_id (message events) or id (room events). An empty string counts as absent at both steps. Unresolvable raises ValueError with one issue on path room_id: code missing when the fallback field is absent or empty, wrong_type when it is present but not a string.

evaluate_next_message(triggering_message_id, next_message_id) compares the triggering message against the platform's authoritative "next open message" for a room (already fetched by the caller): "no_pending", "already_processed", or "ready_to_claim".

evaluate_drain_candidate(candidate, seen_ids, agent_id) classifies one already-fetched drain candidate — candidate is a mapping with id, sender_id, sender_type, or None when the fetch returned nothing: "no_candidate", "self_echo" (checked before the snapshot, so an echo never halts a drain), "out_of_snapshot" (stop), or "drain" (continue). The caller's own bounded or unbounded loop owns the cap; this function classifies one candidate per call.

evaluate_adapter_result(room_id, message_id, succeeded) maps an adapter outcome to "processed" or "failed".

is_self_echo(sender_id, sender_type, agent_id) is the one definition of "self echo" (sender_type == "Agent" and sender_id == agent_id) that evaluate_delivery_event and evaluate_drain_candidate use internally, exposed for a host's own call sites that classify a sender without going through either function.

Turn outcome

TurnLedger judges one agent turn — design decisions live as rustdoc in crates/core/src/turn.rs. Create one per turn, record the TurnEffect of each successful tool call (a failed call records nothing), and ask for the verdict() when the model stops: TurnVerdict.Complete once the turn replied, declined, or did real work, else TurnVerdict.MissingReply (fail the turn with missing_reply_message()). settle() marks a turn the SDK answered itself and note_reported() one whose failure was already reported; both complete it. reply_settled() is True once a recorded call posted to the room or declined to, so no fallback text is due. record takes a TurnEffect only; a wire-name string raises TypeError (convert one with TurnEffect.from_wire_name).

band_tool_effects() maps every canonical Band tool name to its TurnEffect; normalize an alias to its canonical name before the lookup. no_reply_tool() returns band_no_reply's definition as a ToolSpec dict (name, description, parameters).

from band_sdk_core import TurnEffect, TurnLedger, TurnVerdict, band_tool_effects

ledger = TurnLedger()
ledger.record(band_tool_effects()["band_send_event"])
assert ledger.verdict() == TurnVerdict.MissingReply

ledger.record(TurnEffect.Reply)
assert ledger.verdict() == TurnVerdict.Complete
assert ledger.reply_settled()

Values across the language boundary

JSON objects become Python dicts. JSON null becomes None. An explicit null stays distinct from an absent key — a distinction the payload validator uses. Values that cannot be converted raise TypeError, which except Exception catches.

Build / test

just check type-checks and lints this crate. Linking the extension happens through uv / maturin (just test-py, just build-py), not cargo test.

just test-py     # uv sync, pytest, isolated wheel install
just build-py    # uv build (wheel only)

Wheels are platform-specific. Recipes use uv and honor UV_PYTHON, or PYTHON / PYTHON_BIN. Windows builds need the MSVC toolchain rustup's default host target uses.

The wheel is abi3 for Python >= 3.11. This is a mixed Python/Rust maturin package: python/band_sdk_core/ holds the typed surface — __init__.pyi, an empty py.typed, and an __init__.py that re-exports the compiled extension and defines the delivery decisions' TypedDict return shapes (pure typing constructs with no pyo3 equivalent). just test-py checks the stub matches the runtime package and that both ship in the wheel.

License

MIT — see LICENSE.

Fenced delivery settlement

guarded_claim uses the same eligibility decision for every discovery path. Pending originals stay excluded even when parked. Record terminal intent before awaiting transport; call confirm_settlement only after durable acceptance and a current-execution check. An uncertain response leaves the exact intent retained. Retry confirmation restores charged attempts into a replacement tracker without marking the input completed; it keeps the stricter retry ceiling.

from band_sdk_core import ClaimRegistry, RetryTracker

claims = ClaimRegistry()
previous = RetryTracker(max_retries=3)
assert claims.guarded_claim("room", "message", previous) == "runnable"
assert previous.record_attempt("message") == (1, False)
pending = claims.record_failed_settlement("room", "message", "failure", previous, True)
claims.release("room", "message")
current = RetryTracker(max_retries=3)
assert claims.guarded_claim("room", "message", current) == "pending"
assert claims.confirm_settlement("room", "message", pending["token"], current)
assert not claims.is_completed("room", "message")
assert current.record_attempt("message") == (2, False)