Skip to content

symfonic.platform.conversation_pg_pause

conversation_pg_pause

Durable pause records that bind a selected-root token to one conversation turn.

A pause row is written only after the HTTP root has authenticated the token's signed session and claim, and only while the conversation is still at the revision the paused turn prepared. It is the selected root's issuance record: a token with no row here was not paused by this root and is not resumed by it. Resuming publishes the paused query and the final answer in the same transaction that moves the row from pending to published and compare-and-sets the conversation revision, so a resumed answer is published at most once and never over a turn another worker published meanwhile.

A resume that fails after its token was claimed moves the row to abandoned: the token is spent, so nothing can redeem the pause again and the row stops pinning an empty session. A row orphaned by a process that died mid-resume is released by its signed expires_at instead.

PostgresConversationPauses 

PostgresConversationPauses(authority: Any)

Pause verbs over one PostgresConversationAuthority's tables and pool.

Source code in src/symfonic/platform/conversation_pg_pause.py 
49
50
51
def __init__(self, authority: Any) -> None:
    self._authority = authority
    self._tables = authority.tables

abandon async 

abandon(principal: Any, session_id: str, claim_id: str) -> bool

Settle a claimed pause whose resumed turn published nothing.

Call only after this caller won the token's single-use claim; the token is spent, so this never makes the pause redeemable again. Returns whether a pending row was settled.

Source code in src/symfonic/platform/conversation_pg_pause.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
async def abandon(self, principal: Any, session_id: str, claim_id: str) -> bool:
    """Settle a claimed pause whose resumed turn published nothing.

    Call only after this caller won the token's single-use claim; the
    token is spent, so this never makes the pause redeemable again.
    Returns whether a pending row was settled.
    """
    owner = conversation_owner(principal)
    await self._authority.astart()
    async with self._authority.pool.acquire() as conn:
        moved = await conn.execute(
            f"UPDATE {self._tables.pauses} SET state='abandoned', "
            f"settled_at=clock_timestamp() WHERE {_PAUSE} AND state='pending'",
            *session_key(owner, session_id),
            _claim(claim_id),
            owner.principal_id,
        )
    return bool(moved == "UPDATE 1")

pending async 

pending(principal: Any, session_id: str, claim_id: str) -> PendingPause

The owner's pending pause, still resumable at its turn's revision.

Source code in src/symfonic/platform/conversation_pg_pause.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
async def pending(self, principal: Any, session_id: str, claim_id: str) -> PendingPause:
    """The owner's pending pause, still resumable at its turn's revision."""
    owner = conversation_owner(principal)
    await self._authority.astart()
    async with (
        self._authority.pool.acquire() as conn,
        conn.transaction(isolation="repeatable_read", readonly=True),
    ):
        pause = await self._pause(conn, owner, session_id, claim_id, lock=False)
        row = await self._authority._owned(conn, owner, session_id)
    if row["revision"] != pause.base_revision:
        raise ConversationConflict("conversation moved on after the pause")
    return pause

publish async 

publish(principal: Any, session_id: str, claim_id: str, answer: str) -> int

Publish the paused query and resumed answer exactly once.

Source code in src/symfonic/platform/conversation_pg_pause.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
async def publish(self, principal: Any, session_id: str, claim_id: str, answer: str) -> int:
    """Publish the paused query and resumed answer exactly once."""
    if not isinstance(answer, str) or not answer.strip():
        raise ValueError("empty conversation answer")
    owner = conversation_owner(principal)
    await self._authority.astart()
    async with self._authority.pool.acquire() as conn, conn.transaction():
        pause = await self._pause(conn, owner, session_id, claim_id, lock=True)
        revision = await self._authority.append_exchange(
            principal, session_id, pause.base_revision, pause.query, answer, conn=conn
        )
        await conn.execute(
            f"UPDATE {self._tables.pauses} SET state='published', "
            f"settled_at=clock_timestamp() WHERE {_PAUSE} AND state='pending'",
            *session_key(owner, session_id),
            pause.claim_id,
            owner.principal_id,
        )
        return revision

record async 

record(principal: Any, session_id: str, claim_id: str, query: str, base_revision: int, *, expires_at: int, supersedes: str | None = None) -> None

File a pending pause; refuse if the turn's revision has moved on.

expires_at is the token's signed expiry (epoch seconds).

Source code in src/symfonic/platform/conversation_pg_pause.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
async def record(
    self,
    principal: Any,
    session_id: str,
    claim_id: str,
    query: str,
    base_revision: int,
    *,
    expires_at: int,
    supersedes: str | None = None,
) -> None:
    """File a pending pause; refuse if the turn's revision has moved on.

    ``expires_at`` is the token's signed expiry (epoch seconds).
    """
    owner = conversation_owner(principal)
    key = session_key(owner, session_id)
    claim = _claim(claim_id)
    if type(query) is not str or type(base_revision) is not int or type(expires_at) is not int:
        raise ConversationRefused("pause needs the paused query, revision and expiry")
    await self._authority.astart()
    async with self._authority.pool.acquire() as conn, conn.transaction():
        row = await self._authority._owned(conn, owner, session_id, lock=True)
        if row["revision"] != base_revision:
            raise ConversationConflict("conversation revision changed before the pause")
        if supersedes is not None:
            moved = await conn.execute(
                f"UPDATE {self._tables.pauses} SET state='repaused', "
                f"settled_at=clock_timestamp() WHERE {_PAUSE} AND state='pending' "
                "AND base_revision=$7",
                *key,
                _claim(supersedes),
                owner.principal_id,
                base_revision,
            )
            if moved != "UPDATE 1":
                raise ConversationConflict("the resumed pause is no longer pending")
        inserted = await conn.fetchval(
            f"INSERT INTO {self._tables.pauses} (tenant_id, sub_tenant_id, namespace, "
            "session_id, claim_id, principal_id, query, base_revision, expires_at, state, "
            "created_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 'pending', "
            "clock_timestamp()) ON CONFLICT DO NOTHING RETURNING claim_id",
            *key,
            claim,
            owner.principal_id,
            query,
            base_revision,
            expires_at,
        )
        if inserted is None:
            raise ConversationConflict("pause claim already recorded")