Skip to content

symfonic.capabilities.memory.conversation_generation

conversation_generation

Erase generations for a standalone anchored transcript (CORE-171, CORE-199 P3-4).

An exact NativeConversationHistory over an anchored PostgresConversationLog composed with memory that is not raw graph-only PostgreSQL v1 gets no TurnGenerations: nothing memory-side can prove its writes. The transcript still can. Before the turn reads, the history captures the generations covering its conversation from the log's own database; the final append re-proves them under the cursor lock, which an admin forget's advance conflicts with. A turn that read before an acknowledged forget therefore refuses to publish after it.

The capture is per kernel run, keyed weakly on its issued lineage, and the first one is final, as in turn_generation. A run started inside another re-proves every ancestor's capture on the same log before it captures its own, so a child started after a forget of its parent's conversation refuses too. The anchor, not this token, is what keeps a restored database from serving a forgotten transcript; a recovered or caller-supplied transcript refuses here because no persisted generation covers it. Only that exact composition can be fenced, so an anchored log under a subclass of either refuses at binding rather than serving turns no generation re-proves.

ConversationGenerations 

ConversationGenerations(log: Any, *, _mint: object = None)

The transcript token each live run captured; fail-closed, never replaced.

Source code in src/symfonic/capabilities/memory/conversation_generation.py 
51
52
53
54
55
56
def __init__(self, log: Any, *, _mint: object = None) -> None:
    from symfonic.capabilities.memory.turn_generation import UNMINTED

    if _mint is not _MINT:
        raise MemoryContractError(UNMINTED)
    self._log = log

capture async 

capture(path: str, *, recovered: bool = False, history: bool = False) -> None

Observe the log's generations before the turn reads; a carried transcript refuses.

Source code in src/symfonic/capabilities/memory/conversation_generation.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
async def capture(self, path: str, *, recovered: bool = False, history: bool = False) -> None:
    """Observe the log's generations before the turn reads; a carried transcript refuses."""
    from symfonic.capabilities.memory.turn_generation import HISTORY, RECOVERED, _lineage
    from symfonic.memory.backends.postgres_conversation_generation import capture

    if recovered or history:
        raise MemoryContractError(RECOVERED if recovered else HISTORY)
    lineage = _lineage()
    captures = _RUNS.setdefault(lineage, {})
    if self not in captures:
        await self._ancestors_current(lineage)
        token = await capture(self._log.pool, path)
        captures.setdefault(self, (path, token))
    self._own(path)

prove_runs async staticmethod 

prove_runs(runs: tuple[Any, ...]) -> None

The kernel's run-start preflight: each capture of runs on its own log.

Source code in src/symfonic/capabilities/memory/conversation_generation.py 
90
91
92
93
94
95
96
97
@staticmethod
async def prove_runs(runs: tuple[Any, ...]) -> None:
    """The kernel's run-start preflight: each capture of ``runs`` on its own log."""
    from symfonic.memory.backends.postgres_conversation_generation import require

    for run in runs:
        for registry, (path, token) in tuple(_RUNS.get(run, {}).items()):
            await require(registry._log.pool, path, token)  # noqa: SLF001

token 

token(path: str) -> tuple[tuple[str, int], ...]

The run's captured token, which the final append re-proves.

Source code in src/symfonic/capabilities/memory/conversation_generation.py 
107
108
109
def token(self, path: str) -> tuple[tuple[str, int], ...]:
    """The run's captured token, which the final append re-proves."""
    return self._own(path)

bind_conversation 

bind_conversation(conversation: Any) -> Any

conversation with a registry when it is an exact history over an anchored log.

Source code in src/symfonic/capabilities/memory/conversation_generation.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def bind_conversation(conversation: Any) -> Any:
    """``conversation`` with a registry when it is an exact history over an anchored log."""
    from symfonic.capabilities.memory.turn_generation import _exact_history

    if not _exact_history(conversation):
        if _anchored(conversation):
            raise MemoryContractError(INEXACT)
        return conversation
    if conversation.store.proof is None:
        return conversation
    from dataclasses import replace

    return replace(conversation,
                   generations=ConversationGenerations(conversation.store, _mint=_MINT))