symfonic.platform.conversation_authority

conversation_authority

One authenticated owner for opt-in native HTTP checkpoint transcripts.

This composition owns the configured legacy SQLite store. It admits only native chat; legacy invocation and continuation families must refuse at HTTP ingress. Unowned historical checkpoints are never adopted from a caller's session id. Use one authority per saver/writer process; distributed writers use conversation_pg.PostgresConversationAuthority instead.

ConversationAuthority

ConversationAuthority(factory: Any)

Checkpoint-backed owner with final-only, revision-checked publication.

Compose transcript_reader() over this owner's saver for the selected root. compatibility_reader remains for compositions that still inject the facade's reader; an authority holds exactly one reader of either kind. create is the only way to issue new sessions; prepare never registers an unknown id. close belongs to the composition lifetime. The factory must represent an exclusively owned local saver; sharing the saver with another writer is refused when a checkpoint revision changes.

Source code in src/symfonic/platform/conversation_authority.py

def __init__(self, factory: Any) -> None:
    if type(factory) is not LocalConversationFactory:
        raise TypeError("shared conversations require an exclusive local SQLite factory")
    factory.claim()
    self._factory = factory
    try:
        self._saver = factory.build_saver_sync()
    except BaseException:
        factory.release_claim()
        raise
    self._publish_lock = asyncio.Lock()
    self._reader: Any = None

authorize async

authorize(principal: Any, session_id: str) -> None

Refuse exactly as transcript does, without reading the transcript.

Source code in src/symfonic/platform/conversation_authority.py

async def authorize(self, principal: Any, session_id: str) -> None:
    """Refuse exactly as ``transcript`` does, without reading the transcript."""
    await self._owned(principal, session_id)

compatibility_reader

compatibility_reader(factory: TranscriptReaderFactory) -> Any

Compose the existing facade's transcript reader over this same file.

The selected HTTP root refuses every legacy invocation family. This facade supplies the existing public transcript read, not model execution. Legacy configuration belongs to the injected factory, not this owner.

Source code in src/symfonic/platform/conversation_authority.py

def compatibility_reader(self, factory: TranscriptReaderFactory) -> Any:
    """Compose the existing facade's transcript reader over this same file.

    The selected HTTP root refuses every legacy invocation family. This
    facade supplies the existing public transcript read, not model execution.
    Legacy configuration belongs to the injected factory, not this owner.
    """
    if self._reader is not None:
        raise ValueError("conversation authority already has a transcript reader")
    self._reader = factory.create(self._factory.path)
    return self._reader

create async

create(principal: Any) -> str

Issue a new native-family session after verification.

Source code in src/symfonic/platform/conversation_authority.py

async def create(self, principal: Any) -> str:
    """Issue a new native-family session after verification."""
    self._identity(principal)
    session_id = uuid.uuid4().hex
    await self.astart()
    config = self._config(principal, session_id)
    if await self._saver.aget_tuple(config) is not None:
        raise ConversationConflict("session issuance collision")
    await self._write(config, self._owner(principal, session_id), (), 0)
    return session_id

discard_empty async

discard_empty(principal: Any, session_id: str) -> None

Remove a first-turn session only if no completed turn was published.

Source code in src/symfonic/platform/conversation_authority.py

async def discard_empty(self, principal: Any, session_id: str) -> None:
    """Remove a first-turn session only if no completed turn was published."""
    async with self._publish_lock:
        row = await self._owned(principal, session_id)
        if row.checkpoint["channel_values"].get("http_turns", 0) == 0:
            await self._saver.adelete_thread(row.config["configurable"]["thread_id"])

list_sessions async

list_sessions(principal: Any) -> list[dict[str, Any]]

List only this verified principal's native sessions from the saver.

Source code in src/symfonic/platform/conversation_authority.py

async def list_sessions(self, principal: Any) -> list[dict[str, Any]]:
    """List only this verified principal's native sessions from the saver."""
    tenant, sub, user = self._identity(principal)
    namespace = str(getattr(principal.scope, "namespace", None) or "")
    await self.astart()
    latest: dict[str, dict[str, Any]] = {}
    for row in await self._factory.latest_for_owner(tenant, sub, user, namespace):
        if row is None:
            continue
        values = row.checkpoint.get("channel_values", {})
        owner = values.get("http_owner", {})
        if owner.get("namespace") != namespace:
            continue
        if (
            owner.get("tenant_id"),
            owner.get("sub_tenant_id"),
            owner.get("principal_id"),
            owner.get("family"),
        ) != (tenant, sub, user, "native-chat"):
            continue
        sid = owner["session_id"]
        if sid not in latest:
            latest[sid] = {
                "session_id": sid,
                "tenant_id": tenant,
                "created_at": values["http_created_at"],
                "last_active": values["http_updated_at"],
                "message_count": values["http_turns"],
            }
    return sorted(latest.values(), key=lambda row: row["last_active"], reverse=True)

prepare async

prepare(principal: Any, session_id: str, query: str) -> PreparedConversation

Read authorized history; publish only a usable completed exchange.

Source code in src/symfonic/platform/conversation_authority.py

async def prepare(self, principal: Any, session_id: str, query: str) -> PreparedConversation:
    """Read authorized history; publish only a usable completed exchange."""
    from symfonic import Message

    row = await self._owned(principal, session_id)
    checkpoint = row.checkpoint
    messages = tuple(checkpoint["channel_values"].get("messages", ()))
    history = tuple(
        Message(
            role=m.type.replace("human", "user").replace("ai", "assistant"), content=m.content
        )
        for m in messages
    )
    published = False

    async def publish(answer: str) -> None:
        nonlocal published
        from langchain_core.messages import AIMessage, HumanMessage

        if not isinstance(answer, str) or not answer.strip():
            raise ValueError("empty conversation answer")
        async with self._publish_lock:
            if published:
                return
            current = await self._owned(principal, session_id)
            if current.checkpoint["id"] != checkpoint["id"]:
                raise ConversationConflict("conversation revision changed")
            updated = (
                *messages,
                HumanMessage(content=query, id=uuid.uuid4().hex),
                AIMessage(content=answer, id=uuid.uuid4().hex),
            )
            count = checkpoint["channel_values"].get("http_turns", 0) + 1
            recorded = checkpoint["channel_values"].get(INTRODUCED_AT)
            await self._write(
                current.config,
                self._owner(principal, session_id),
                updated,
                count,
                created_at=checkpoint["channel_values"]["http_created_at"],
                introduced=recorded if isinstance(recorded, dict) else None,
                fresh=(updated[-2].id, updated[-1].id),
            )
            published = True

    return PreparedConversation(history=history, publish=publish)

transcript async

transcript(principal: Any, session_id: str) -> list[Any]

Authorize, then read; the native reader decodes the authorized row.

Source code in src/symfonic/platform/conversation_authority.py

async def transcript(self, principal: Any, session_id: str) -> list[Any]:
    """Authorize, then read; the native reader decodes the authorized row."""
    row = await self._owned(principal, session_id)
    if isinstance(self._reader, SqliteTranscriptReader):
        return await self._reader.read_owned(row)
    if self._reader is None:
        raise ConversationRefused("transcript reader not composed")
    return await self._reader.get_transcript(principal.scope, session_id)

transcript_reader

transcript_reader() -> SqliteTranscriptReader

The native reader over this owner's saver; no facade is constructed.

Source code in src/symfonic/platform/conversation_authority.py

def transcript_reader(self) -> SqliteTranscriptReader:
    """The native reader over this owner's saver; no facade is constructed."""
    if self._reader is not None:
        raise ValueError("conversation authority already has a transcript reader")
    self._reader = SqliteTranscriptReader(self._saver, self.astart)
    return self._reader

verify_principal classmethod

verify_principal(principal: Any) -> None

Refuse any identity that did not originate in a principal verifier.

Source code in src/symfonic/platform/conversation_authority.py

@classmethod
def verify_principal(cls, principal: Any) -> None:
    """Refuse any identity that did not originate in a principal verifier."""
    cls._identity(principal)

ConversationConflict

Bases: SymfonicError

The transcript changed while this request was running.

ConversationFamilyRefused

Bases: SymfonicError

An invocation family has no bridge into the selected native authority.