symfonic.capabilities.memory.working

working

The conversation window: what was just said, before anything is ranked.

This is the one part of hydration the scorer never touches, and that is the point. The previous few turns are relevant because they are recent, not because they resemble the current message — run them through a similarity gate and a user who changes the subject loses the sentence they just typed, which is the exact moment they most expect the assistant to remember it.

So the window is recency-anchored and ungated: no similarity floor, no salience floor, no ranking. The only things it drops are the ones that would corrupt the block rather than merely fill it — an empty turn, and a speaker the deployment excluded.

Speaker exclusion is forward-looking. A turn that declares no speaker is never excluded. The knob exists so an operator can stop the model re-priming on its own prior answers; applying it to rows that predate the field would silently delete a conversation's whole history the day the knob was set.

One turn renders on one line. The [working] prefix is a delimiter, so stored text is flattened before interpolation — otherwise a turn containing a newline and a forged prefix becomes two turns, one of them attributed to a speaker who never said it.

ConversationSource

Bases: Protocol

The working layer, as hydration needs it.

recent async

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

The last limit turns at scope, oldest first.

Raises :class:~.errors.MemoryUnavailable when the working store cannot be reached; :class:WorkingWindow degrades rather than failing the turn.

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

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

    Raises :class:`~.errors.MemoryUnavailable` when the working store cannot
    be reached; :class:`WorkingWindow` degrades rather than failing the turn.
    """
    ...

ConversationTurn dataclass

ConversationTurn(turn_id: str, speaker: str = '', text: str = '', turn: int = 0)

One thing that was said, as the working layer holds it.

line

line() -> str

The rendered form: layer prefix, speaker, single-line text.

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

def line(self) -> str:
    """The rendered form: layer prefix, speaker, single-line text."""
    body = flatten(self.text).strip()
    prefix = f"[{MemoryLayer.WORKING.value}]"
    return f"{prefix} {self.speaker}: {body}" if self.speaker else f"{prefix} {body}"

WorkingContext dataclass

WorkingContext(turns: tuple[ConversationTurn, ...] = (), dropped: tuple[tuple[str, str], ...] = (), degraded: bool = False)

The conversation window as it will render, and what it left out.

render

render() -> str

One turn per line, oldest first.

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

def render(self) -> str:
    """One turn per line, oldest first."""
    return "\n".join(turn.line() for turn in self.turns)

WorkingWindow dataclass

WorkingWindow(source: ConversationSource, recent_turns: int = 0, exclude_speakers: frozenset[str] = frozenset())

Reads the last few turns of a conversation, ungated.

read async

read(scope: MemoryScope) -> WorkingContext

Read the window at scope, dropping only what would corrupt it.

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

async def read(self, scope: MemoryScope) -> WorkingContext:
    """Read the window at ``scope``, dropping only what would corrupt it."""
    if self.recent_turns == 0:
        return WorkingContext()
    try:
        recent = await self.source.recent(scope, self.recent_turns)
    except MemoryUnavailable:
        return WorkingContext(degraded=True)

    state = _Window()
    for turn in recent[-self.recent_turns :]:
        if not turn.text.strip():
            # An empty row renders as a bare "[working] " bullet, which reads
            # as a turn in which nobody said anything.
            state.dropped.append((turn.turn_id, "the turn carries no text"))
            continue
        if turn.speaker and turn.speaker in self.exclude_speakers:
            state.dropped.append(
                (turn.turn_id, f"speaker {turn.speaker!r} is excluded")
            )
            continue
        state.kept.append(turn)
    return WorkingContext(turns=tuple(state.kept), dropped=tuple(state.dropped))