Skip to content

symfonic.platform.conversation_local

conversation_local

Exclusive local composition of the existing SQLite checkpointer factory.

The shared HTTP authority has no distributed checkpoint CAS. This adapter therefore admits one local owner, using a process lease before any saver I/O. Production PostgreSQL/Mongo compositions need their own atomic adapter.

LocalConversationFactory 

LocalConversationFactory(path: str)

Own one configured SQLite saver and its local writer lease.

Source code in src/symfonic/platform/conversation_local.py 
27
28
29
30
31
32
33
34
35
36
37
38
def __init__(self, path: str) -> None:
    if path == ":memory:":
        raise ValueError("shared conversations require a durable SQLite path")
    from symfonic.agent.checkpointer.factory import SqliteCheckpointerFactory

    self._path = Path(path).resolve()
    self._factory = SqliteCheckpointerFactory(str(self._path))
    self._lock = asyncio.Lock()
    self._lease: Any = None
    self._claimed = False
    self._closed = False
    self._pid = os.getpid()

path property 

path: str

Existing legacy configuration's durable SQLite location.

latest_for_owner async 

latest_for_owner(tenant: str, sub: str, principal: str, namespace: str) -> list[Any]

Read one latest snapshot per owned thread, without decoding older history.

This adapter owns the existing SQLite saver schema. Metadata filtering happens in SQL; it never materializes another principal's transcript.

Source code in src/symfonic/platform/conversation_local.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
async def latest_for_owner(
    self,
    tenant: str,
    sub: str,
    principal: str,
    namespace: str,
) -> list[Any]:
    """Read one latest snapshot per owned thread, without decoding older history.

    This adapter owns the existing SQLite saver schema. Metadata filtering
    happens in SQL; it never materializes another principal's transcript.
    """
    await self.astart()
    saver = self.build_saver_sync()
    await saver.setup()
    async with (
        saver.lock,
        saver.conn.execute(
            "SELECT DISTINCT thread_id FROM checkpoints WHERE checkpoint_ns = '' "
            "AND json_extract(CAST(metadata AS TEXT), '$.http_native') = 1 "
            "AND json_extract(CAST(metadata AS TEXT), '$.http_tenant') = ? "
            "AND json_extract(CAST(metadata AS TEXT), '$.http_sub') = ? "
            "AND json_extract(CAST(metadata AS TEXT), '$.http_principal') = ? "
            "AND json_extract(CAST(metadata AS TEXT), '$.http_namespace') = ?",
            (tenant, sub, principal, namespace),
        ) as cursor,
    ):
        threads = await cursor.fetchall()
    return [
        await saver.aget_tuple(
            {
                "configurable": {
                    "thread_id": thread[0],
                    "checkpoint_ns": "",
                }
            }
        )
        for thread in threads
    ]

release_claim 

release_claim() -> None

Release a claim when zero-I/O saver construction fails.

Source code in src/symfonic/platform/conversation_local.py 
62
63
64
65
66
67
def release_claim(self) -> None:
    """Release a claim when zero-I/O saver construction fails."""
    if self._lease is not None:
        self._lease.close()
        self._lease = None
    self._claimed = False

LocalConversationUnavailable

Bases: SymfonicError

The selected local writer cannot provide exclusive authority.