Skip to content

symfonic.memory.backends.postgres_accepted_cadence

postgres_accepted_cadence

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

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

PostgresAcceptedCadence 

PostgresAcceptedCadence(journal: MongoOperationJournal, coordinator: ConsolidationCoordinator, *, policy_digest: str | None = None)

Exact v1 QUICK only; public Agent receipt admission remains closed.

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_accepted_cadence.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def __init__(self, journal: MongoOperationJournal, coordinator: ConsolidationCoordinator,
             *, policy_digest: str | None = None):
    if (type(journal) is not MongoOperationJournal
            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()
    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._journal, self._coordinator = 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 Mongo acceptance, then atomically advance and reserve due QUICK.

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

    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_accepted_cadence.py 
128
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
154
async def recover_scope(self, scope: MemoryScope, *, limit: int = 500):
    """Discover accepted turns without invoking their model or stage bodies."""
    from symfonic.memory.backends.mongo_accepted_inbox import accepted_batch

    if type(scope) is not MemoryScope or scope.identity_version != 1:
        _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_accepted_cadence "
            "WHERE tenant_id=$1 AND scope_path=$2", scope.tenant, scope.path)
        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 accepted_batch(self._journal, scope, after, limit=limit)
    for completion in completions:
        await self.advance(completion)
    return len(completions), more