angee.integrate.live

Console-safe live bridge session facts.

The web/console import path needs pairing vocabulary and store management, but must not load worker-only dependencies. Keep this module free of vendor SDKs and QR rendering libraries; integrate.session owns the worker-only qrcode import.

WAKE_SECONDS

Upper bound between live-session desired-state / shutdown / lock checks.

AWAITING_PASSWORD_WAKE_SECONDS

Deliberate 1 Hz wake while awaiting a password; bounded by operator action.

STOP_JOIN_SECONDS

How long a stopping live session waits for the vendor connection to unwind.

SESSION_EXIT_TIMEOUT

Default destructive-reset wait: one wake, one vendor unwind, and headroom.

This bound derives from the real loop constants so a destructive reset cannot silently drift below the session's maximum stop and unwind time.

skipped_password_marker

def skipped_password_marker(material_key: str) -> dict[str, str]

Return the non-secret persisted marker for a skipped material-key round.

armed_material_key

def armed_material_key(state: object) -> str

Return the material key from an armed awaiting-password state, or "".

is_skip_marker

def is_skip_marker(value: object, material_key: str) -> bool

Return whether value records an explicit skip for material_key.

PairingState

@strawberry.enum
class PairingState(enum.StrEnum)

The pairing vocabulary a live session reports and a connect dialog renders.

from_report

@classmethod
def from_report(cls, value: object) -> PairingState | None

Return the member one serialized report carries, or None.

PairingProjection

@strawberry.type(name="Pairing")
class PairingProjection()

Vendor-neutral pairing state projected by a live bridge implementation.

SessionLoggedOut

class SessionLoggedOut(Exception)

The linked account removed this live session and needs explicit reset.

session_store_path

def session_store_path(bridge: Any) -> Path

Return the bridge's session directory under the stack-persisted data dir.

reset_session_store

def reset_session_store(bridge: Any) -> None

Delete the bridge's session store after the caller proves it is released.

await_session_exit

def await_session_exit(bridge: Any, *, timeout: float | None = None) -> None

Wait for the bridge's advisory lock to clear before touching its store.