Gateway Session Sharing

from praisonaiagents import Agent

agent = Agent(name="assistant", instructions="Be helpful.")
agent.start("Kick off a shared session")
# A teammate attaches from another client to watch the same live turn stream.

from praisonaiagents.gateway import (
    SessionVisibility,
    SessionSharingRole,
    SessionObserverProtocol,
)

from praisonaiagents.gateway import SessionVisibility, SessionSharingRole

# Alice owns the session on her laptop
owner_role = SessionSharingRole.OWNER

# Alice opens the same session read-only on her phone
alice_phone = (SessionVisibility.READ_ONLY, SessionSharingRole.VIEWER)

# Bob joins to co-drive — the session is shared
bob = (SessionVisibility.SHARED, SessionSharingRole.MEMBER)

print(owner_role.value)     # "owner"
print(alice_phone[1].value) # "viewer"
print(bob[1].value)         # "member"

from typing import List, Tuple
from praisonaiagents.gateway import SessionSharingRole, SessionObserverProtocol


class InMemoryObservers:
    def __init__(self) -> None:
        self._by_session: dict[str, dict[str, SessionSharingRole]] = {}

    def attach(
        self,
        session_id: str,
        client_id: str,
        role: SessionSharingRole = SessionSharingRole.VIEWER,
    ) -> None:
        self._by_session.setdefault(session_id, {})[client_id] = role

    def detach(self, session_id: str, client_id: str) -> None:
        self._by_session.get(session_id, {}).pop(client_id, None)

    def observers(
        self, session_id: str,
    ) -> "List[Tuple[str, SessionSharingRole]]":
        return list(self._by_session.get(session_id, {}).items())


impl = InMemoryObservers()
impl.attach("s1", "alice", SessionSharingRole.OWNER)
impl.attach("s1", "bob")  # defaults to VIEWER

print(impl.observers("s1"))
# [('alice', <SessionSharingRole.OWNER>), ('bob', <SessionSharingRole.VIEWER>)]

# Protocol is runtime-checkable
print(isinstance(impl, SessionObserverProtocol))  # True