Skip to content

symfonic.core.contracts.conversation_reader

conversation_reader

Read-only transcript composition port shared by platform and legacy adapter.

ConversationRefused

Bases: SymfonicError

No verified ownership, or an unsupported invocation family.

TranscriptReadRefused 

TranscriptReadRefused()

Bases: ConversationRefused

Versioned direct-reader refusal without revealing checkpoint existence.

Source code in src/symfonic/core/contracts/conversation_reader.py 
30
31
32
33
34
35
36
def __init__(self) -> None:
    super().__init__(
        "Direct transcript read unavailable (direct_transcript_read.v1; 11.4). "
        "Use Agent/AgentHost with ConversationAuthority for native conversations; "
        "legacy checkpoint reads require symfonic-core==9.12.*. "
        "See docs/guides/24-11-transition.md."
    )

TranscriptReader

Bases: Protocol

Compatibility read projection; no invocation or configuration surface.

TranscriptReaderFactory

Bases: Protocol

A composition-owned reader for the authority's selected durable file.

owner_scope 

owner_scope(scope: Any) -> tuple[str, str, str]

The (tenant, sub-tenant, namespace) a conversation is filed under.

Refuses a scope whose key would be ambiguous: no tenant, the literal absent-sub placeholder "_" as a sub-tenant, or a ":" in either, since the thread key joins them with ":".

Source code in src/symfonic/core/contracts/conversation_reader.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def owner_scope(scope: Any) -> tuple[str, str, str]:
    """The ``(tenant, sub-tenant, namespace)`` a conversation is filed under.

    Refuses a scope whose key would be ambiguous: no tenant, the literal
    absent-sub placeholder ``"_"`` as a sub-tenant, or a ``":"`` in either,
    since the thread key joins them with ``":"``.
    """
    if not getattr(scope, "tenant_id", None) or getattr(scope, "sub_tenant_id", None) == "_":
        raise ConversationRefused("ambiguous conversation scope")
    tenant = str(scope.tenant_id)
    sub = str(getattr(scope, "sub_tenant_id", None) or "_")
    if ":" in tenant or ":" in sub:
        raise ConversationRefused("ambiguous conversation scope")
    return tenant, sub, str(getattr(scope, "namespace", None) or "")