Skip to content

symfonic.platform.working_pg_create

working_pg_create

Transactional opt-in Working POST; TTL requires offline DB authority.

PostgresWorkingCreateService 

PostgresWorkingCreateService(reference, graph, store, guard, owner, *, retention, sink)
Source code in src/symfonic/platform/working_pg_create.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def __init__(self, reference, graph, store, guard, owner, *, retention, sink):
    if (
        type(graph) is not PostgresGraphBackend
        or type(owner) is not PostgresWorkingOwner
        or type(store) is not GraphBackedHms or store.graph is not graph
        or graph.pool is not guard.pool or owner.pool is not graph.pool
    ):
        raise TypeError("Working POST requires the exact PostgreSQL Agent graph")
    if type(retention) is not int or retention < 1:
        raise ValueError("Working retention must be positive")
    if sink is None or isinstance(sink, NullAuditSink):
        raise AuditSinkError("Working POST requires an explicit audit sink")
    self.reference, self.graph, self.store = reference, graph, store
    self.guard, self.owner = guard, owner
    self.retention, self.audit = retention, AuditSeam(sink)
    self._audit_wait_degraded = False

reconcile async 

reconcile(caller, session_id, record_id)

Read current committed owner-visible state; not proof of noncommit.

Source code in src/symfonic/platform/working_pg_create.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
async def reconcile(self, caller, session_id, record_id):
    """Read current committed owner-visible state; not proof of noncommit."""
    if bound_connection(self.graph.pool) is not None:
        raise WorkingWriteUnavailable("Working reconcile refuses ambient transaction")
    await self.owner.check(caller, session_id)
    scope = memory_scope(caller.scope)
    row = await self.graph.get_node(tenant_scope(scope), NodeId(record_id))
    visible = bool(
        row is not None and row.layer == MemoryLayer.WORKING
        and row.properties.get("scope_path") == tenant_scope(scope).scope_path
        and row.properties.get("session_id") == session_id
        and row.properties.get(_POST_MARKER) is True
        and await self._registered(tenant_scope(scope), record_id)
        and not is_pending(row.properties) and not is_retracted(row.properties)
    )
    return {"id": record_id, "state": "committed" if visible else "not_found"}