Skip to content

symfonic.capabilities.memory.window_overlap

window_overlap

Recall the conversation window already renders is not rendered twice.

A turn is written to the transcript and to EPISODIC under one turn_id (the run id) and one speaker; every shipped window source reads the same pair back. When the window renders a turn, its recalled episodic copy is the same utterance, and the model read the last exchange twice.

Identity is exact, never textual: a memory that says the same words on another turn, the other side of a window turn, or a classified or semantic memory that happens to carry a turn tag is a distinct memory and stays. Only transcript rows (:func:~.record_admission.conversation_local) are candidates.

Only a window line the block actually admits suppresses anything. The window leads the block and admission is greedy over that prefix, so the same ceiling over the window lines alone (:func:~.rendered_selection.admitted_mask) says which turns render. A window line the ceiling refuses leaves its recalled copy in place; suppressing it too would show the model neither.

Filtering happens after selection, so ranking, activation seeds and the retrieval result itself are untouched; only the rendered block changes.

outside_window 

outside_window(result: Any, turns: Iterable[Any], *, char_ceiling: int, max_bytes: int | None = None) -> tuple[tuple, tuple, tuple]

Return (memories, linked, dropped) without the window's rendered turns.

Source code in src/symfonic/capabilities/memory/window_overlap.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def outside_window(
    result: Any, turns: Iterable[Any], *, char_ceiling: int, max_bytes: int | None = None,
) -> tuple[tuple, tuple, tuple]:
    """Return ``(memories, linked, dropped)`` without the window's rendered turns."""
    from symfonic.capabilities.memory.rendered_selection import admitted_mask

    turns = tuple(turns)
    rendered = admitted_mask([turn.line() for turn in turns],
                             char_ceiling=char_ceiling, max_bytes=max_bytes)
    window = {
        (turn.turn_id, turn.speaker)
        for turn, fits in zip(turns, rendered, strict=True)
        if fits and turn.turn_id and turn.speaker
    }
    if not window:
        return result.memories, result.linked, ()
    from symfonic.capabilities.memory.record_admission import conversation_local

    def repeats(record: Any) -> bool:
        identity = (record.metadata.get("turn_id"), record.metadata.get("speaker"))
        return identity in window and conversation_local(record)

    memories = tuple(item for item in result.memories if not repeats(item.record))
    linked = tuple(item for item in result.linked if not repeats(item.record))
    dropped = tuple(
        (item.record.record_id, REPEATED_REASON)
        for item in (*result.memories, *result.linked) if repeats(item.record)
    )
    return memories, linked, dropped