Skip to content

symfonic.services.conversation.transcript

transcript

Transcript persistence and query.

Surface separation: this is verbatim replay, not episodic memory. The source is a port, so the same query semantics hold over a checkpointer, an adopter's chat store, or a test double.

The semantics that must not drift from the legacy surface:

  • index is the ordinal within the speaker-filtered view (so "the first thing I asked" is index 0 of a user-filtered read), and it is selectable — TranscriptQuery.index, negatives counting from the end, out-of-range answering empty — because the legacy read offered it and porting a caller to "fetch the thread and slice client-side" is a behaviour change on every large thread;
  • timestamp is checkpoint granularity and is None when the source cannot resolve one — benign for ordinal reads, fatal for a time-range read, which is why the time-range path raises instead of returning everything; and
  • limit is a cap applied last, after the ordinal/time filtering, and it keeps the first rows of that view — the legacy rows[:limit]. A tail window would answer a different question for an identical query during the window in which both paths must be interchangeable.

TranscriptQuery dataclass 

TranscriptQuery(thread_id: str, speaker: Speaker = 'all', limit: int | None = None, index: int | None = None, since: datetime | None = None, until: datetime | None = None)

One transcript read. Validated at construction, not at the store.

TranscriptService 

TranscriptService(*, source: TranscriptSourcePort)

Reads verbatim transcripts through a source port.

Source code in src/symfonic/services/conversation/transcript.py 
111
112
def __init__(self, *, source: TranscriptSourcePort) -> None:
    self._source = source

read async 

read(query: TranscriptQuery) -> tuple[TranscriptRow, ...]

Filter, then select an ordinal, then cap — the legacy order.

limit is applied last and keeps the first rows of the resulting view (rows[:limit]), exactly as the legacy read did. Rows keep the ordinal of the speaker-filtered view they came from, so a capped or ordinal-selected read still correlates with an uncapped one.

Source code in src/symfonic/services/conversation/transcript.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
async def read(self, query: TranscriptQuery) -> tuple[TranscriptRow, ...]:
    """Filter, then select an ordinal, then cap — the legacy order.

    ``limit`` is applied last and keeps the *first* rows of the resulting
    view (``rows[:limit]``), exactly as the legacy read did. Rows keep the
    ordinal of the speaker-filtered view they came from, so a capped or
    ordinal-selected read still correlates with an uncapped one.
    """
    messages = await self._source.messages(query.thread_id)
    stamps = await self._resolve_stamps(query)
    rows = self._project(messages, query, stamps)
    if query.index is not None:
        rows = _select_ordinal(rows, query.index)
    if query.limit is not None:
        rows = rows[: query.limit]
    return rows

TranscriptSourcePort

Bases: Protocol

Where verbatim turns are read from.

messages async 

messages(thread_id: str) -> Any

Return the thread's messages, oldest first.

Source code in src/symfonic/services/conversation/transcript.py 
41
42
async def messages(self, thread_id: str) -> Any:
    """Return the thread's messages, oldest first."""

timestamps async 

timestamps(thread_id: str) -> dict[str, datetime] | None

Map message id -> introducing checkpoint time, or None.

Source code in src/symfonic/services/conversation/transcript.py 
44
45
async def timestamps(self, thread_id: str) -> dict[str, datetime] | None:
    """Map message id -> introducing checkpoint time, or ``None``."""