Skip to content

symfonic.platform.working_create

working_create

Committed Working creation on one exact, shared InMemory reference graph.

No FIFO mirror, scope-wide flush, PostgreSQL admission or default route. Unknown postcommit outcomes carry a reconciliation id, never a false 201.

WorkingCreateService 

WorkingCreateService(reference, graph, store, authority, *, retention, sink)
Source code in src/symfonic/platform/working_create.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(self, reference, graph, store, authority, *, retention, sink):
    if (
        type(graph) is not CommittedInMemoryGraph
        or type(store) is not GraphBackedHms
        or store.graph is not graph
        or type(authority) is not InMemoryWorkingWriteAuthority
        or authority.graph is not graph
    ):
        raise TypeError("Working create requires the exact shared InMemory committed graph")
    if type(retention) is not int or retention < 1:
        raise ValueError("Working retention must be a positive integer")
    if sink is None or isinstance(sink, NullAuditSink):
        raise AuditSinkError("Working creation requires an explicit audit sink")
    self.reference, self.graph, self.store, self.authority = reference, graph, store, authority
    self.retention, self.audit = retention, AuditSeam(sink)
    self._audit_wait_degraded = False

reconcile async 

reconcile(caller, session_id, record_id)

In-process operator recovery, not HTTP. Never automatically retry.

not_found means no currently visible matching row, NOT that the write never committed: retention/retraction can remove it afterwards.

Source code in src/symfonic/platform/working_create.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
async def reconcile(self, caller, session_id, record_id):
    """In-process operator recovery, not HTTP. Never automatically retry.

    ``not_found`` means no currently visible matching row, NOT that the
    write never committed: retention/retraction can remove it afterwards.
    """
    scope = memory_scope(caller.scope)
    async with self.authority.hold(caller, session_id), self.graph.committed_read():
        row = await self.graph.get_node(tenant_scope(scope), record_id)
        committed = 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 not is_pending(row.properties)
            and not is_retracted(row.properties)
        )
    return {"id": record_id, "state": "committed" if committed else "not_found"}