Skip to content

symfonic.memory.backends.postgres_native_accepted_cadence

postgres_native_accepted_cadence

Private native accepted-turn cadence joined to PostgreSQL QUICK receipts.

PostgreSQL acceptance precedes the PostgreSQL cursor/receipt transaction. A failed handoff is UNKNOWN; recovery rereads acceptance without replaying the turn body.

PostgresNativeAcceptedCadence 

PostgresNativeAcceptedCadence(journal: PostgresAcceptedHandoff, coordinator: ConsolidationCoordinator, *, policy_digest: str | None = None)

Exact v1 QUICK only, bound to a factory-issued PostgreSQL handoff.

The caller pins the complete phase configuration (not merely phase names) with a stable lowercase SHA-256 digest across worker restarts.

Source code in src/symfonic/memory/backends/postgres_native_accepted_cadence.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def __init__(self, journal: PostgresAcceptedHandoff, coordinator: ConsolidationCoordinator,
             *, policy_digest: str | None = None):
    if (type(journal) is not PostgresAcceptedHandoff
            or type(coordinator) is not ConsolidationCoordinator
            or type(policy_digest) is not str or len(policy_digest) != 64
            or any(character not in "0123456789abcdef" for character in policy_digest)):
        _refuse()
    if journal._domain is not coordinator._domain:
        _refuse()
    domain, leases = coordinator._domain, coordinator.leases
    if (type(domain) is not PostgresReceiptDomain or type(leases) is not PostgresReceiptLeases
            or leases.transaction_domain is not domain or coordinator.background
            or not coordinator._runtime.native_exact_owner
            or coordinator._cadence._cycles != (ConsolidationCycle.QUICK,)):
        _refuse()
    require_graph_only(coordinator._runtime, domain)
    self._handoff, self._scope = journal, journal._scope
    self._journal, self._coordinator = journal._journal, coordinator
    self._domain, self._leases, self._pool = domain, leases, domain.pool
    schedule = coordinator.schedule
    policy = [policy_digest, schedule.quick_every_turns, schedule.nightly_after_seconds,
              schedule.deep_after_seconds, coordinator.registered]
    self._policy = hashlib.sha256(json.dumps(policy, sort_keys=True).encode()).hexdigest()

advance async 

advance(completion: CompletedTurn)

Reread PostgreSQL acceptance, then atomically advance and reserve due QUICK.

Source code in src/symfonic/memory/backends/postgres_native_accepted_cadence.py 
155
156
157
158
159
160
161
async def advance(self, completion: CompletedTurn):
    """Reread PostgreSQL acceptance, then atomically advance and reserve due QUICK."""
    from symfonic.memory.backends.postgres_native_accepted_execution import advance_accepted

    if type(completion) is not CompletedTurn or completion.full_scope != self._scope:
        _refuse()
    return await advance_accepted(self, completion)

recover_scope async 

recover_scope(scope: MemoryScope, *, limit: int = 500)

Discover accepted turns without invoking their model or stage bodies.

Source code in src/symfonic/memory/backends/postgres_native_accepted_cadence.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
async def recover_scope(self, scope: MemoryScope, *, limit: int = 500):
    """Discover accepted turns without invoking their model or stage bodies."""
    if type(scope) is not MemoryScope or scope != self._scope:
        _refuse()
    conn = await self._pool.acquire()
    try:
        row = await conn.fetchrow(
            "SELECT policy_digest, last_acceptance_sequence, pending_operation_id, "
            "pending_intent_digest FROM memory_pg_accepted_cadence "
            "WHERE tenant_id=$1 AND scope_key=$2", scope.tenant, _key(scope))
        if row is not None:
            if row["policy_digest"] != self._policy:
                _refuse()
            if (row["pending_operation_id"] is not None
                    and not await self._committed(
                        conn, scope, row["pending_operation_id"],
                        row["pending_intent_digest"])):
                _unknown()
        after = 0 if row is None else row["last_acceptance_sequence"]
    finally:
        await self._pool.release(conn)
    completions, more = await self._handoff.recover(after, limit=limit)
    for completion in completions:
        await self.advance(completion)
    return len(completions), more