symfonic.capabilities.memory.conversation

conversation

Reading legacy's working layer as conversation turns (design §2c).

WorkingWindow needs a :class:~.working.ConversationSource; legacy keeps the conversation as MemoryEntry rows in its working layer. The design's rule for this seam is that a turn's identity, speaker and order must not be fabricated out of those rows — and reading the producer showed none of the three has to be:

• turn_id and speaker are framework-reserved metadata keys (engine._RESERVED_AUTO_TURN_METADATA_KEYS), written on every auto-turn write, and an adopter cannot override them, so they carry framework meaning;
• rehydrate_from_messages writes the same shape on purpose, so a replayed row is indistinguishable from a live one;
• get_context documents chronological order, oldest first.

Identity and speaker are therefore read, and order is what the producer states by the sequence it returns. What this module must never do is supply one of them when the producer did not — see :meth:LegacyConversationSource.recent.

LegacyConversationSource

LegacyConversationSource(layer: Any)

A :class:~.working.ConversationSource over legacy's working layer.

Holds the layer rather than reaching for a global, so two agents in one process read two conversations.

Source code in src/symfonic/capabilities/memory/conversation.py

def __init__(self, layer: Any) -> None:
    self._layer = layer

recent async

recent(scope: MemoryScope, limit: int) -> Sequence[ConversationTurn]

The last limit turns at scope, oldest first.

Rows with no turn_id are excluded, not repaired. The obvious repair — fall back to entry.id — is wrong in a way nothing downstream could detect: the engine writes the user utterance and the assistant reply under the same turn_id, so row ids would split one turn into two, and the rendered block would look exactly the same.

A missing speaker is left empty instead, because the window's own rule is that a turn declaring no speaker is never excluded — applying speaker exclusion to rows that predate the field would delete a conversation's history the day an operator set the knob.

turn is the row's position in what the producer returned. Positions are not renumbered after an exclusion: renumbering would turn turn into a count of admitted rows rather than a place in the conversation.

Source code in src/symfonic/capabilities/memory/conversation.py

async def recent(
    self, scope: MemoryScope, limit: int
) -> Sequence[ConversationTurn]:
    """The last ``limit`` turns at ``scope``, oldest first.

    Rows with no ``turn_id`` are **excluded, not repaired**. The obvious
    repair — fall back to ``entry.id`` — is wrong in a way nothing
    downstream could detect: the engine writes the user utterance and the
    assistant reply under the *same* ``turn_id``, so row ids would split one
    turn into two, and the rendered block would look exactly the same.

    A missing ``speaker`` is left empty instead, because the window's own
    rule is that a turn declaring no speaker is never excluded — applying
    speaker exclusion to rows that predate the field would delete a
    conversation's history the day an operator set the knob.

    ``turn`` is the row's position in what the producer returned. Positions
    are not renumbered after an exclusion: renumbering would turn ``turn``
    into a count of admitted rows rather than a place in the conversation.
    """
    if limit <= 0:
        return ()
    try:
        entries = await self._layer.get_context(_legacy_scope(scope))
    except MemoryUnavailable:
        raise
    except Exception as exc:  # noqa: BLE001 - one contract for one condition
        # ``WorkingWindow`` degrades on ``MemoryUnavailable`` and on nothing
        # else, so a backend's own error class must be translated here or
        # the window fails the turn instead of dropping its block.
        raise MemoryUnavailable(
            f"the legacy working layer could not be read: {exc}"
        ) from exc

    turns: list[ConversationTurn] = []
    for position, entry in enumerate(entries or ()):
        metadata = dict(getattr(entry, "metadata", None) or {})
        turn_id = metadata.get(TURN_ID_KEY)
        if not isinstance(turn_id, str) or not turn_id:
            continue
        turns.append(
            ConversationTurn(
                turn_id=turn_id,
                speaker=str(metadata.get(SPEAKER_KEY) or ""),
                text=str(getattr(entry, "content", "")),
                turn=position,
            )
        )
    # Sliced after admission so ``limit`` counts turns the window will
    # render, not rows the store happened to hold.
    return tuple(turns[-limit:])