Skip to content

symfonic.memory.backends.in_memory_ownership

in_memory_ownership

Ownership guards for mixed-version reference stores, loaded on demand.

check_edge_endpoints 

check_edge_endpoints(store: Any, scope: Any, edge: Any) -> None

Native edges require two live endpoints on the exact ancestor chain.

Source code in src/symfonic/memory/backends/in_memory_ownership.py 
170
171
172
173
174
175
176
177
178
179
180
181
def check_edge_endpoints(store: Any, scope: Any, edge: Any) -> None:
    """Native edges require two live endpoints on the exact ancestor chain."""
    if not _is_native_scope(scope):
        return
    from symfonic.memory.pending import is_pending
    from symfonic.memory.retraction import is_retracted
    nodes = store._nodes.get(scope.tenant_id, {})
    for identifier in (edge.source, edge.target):
        node = nodes.get(str(identifier))
        if (node is None or not store._node_visible(scope, node)
            or is_pending(node.properties) or is_retracted(node.properties)):
            raise ValueError("Native edge endpoints must be live and visible in this scope")

insert_new_node 

insert_new_node(store: Any, scope: Any, node: Any) -> Any

Insert only when the tenant-global ID is vacant; caller holds the writer lock.

Source code in src/symfonic/memory/backends/in_memory_ownership.py 
156
157
158
159
160
161
162
163
164
165
166
167
def insert_new_node(store: Any, scope: Any, node: Any) -> Any:
    """Insert only when the tenant-global ID is vacant; caller holds the writer lock."""
    from symfonic.memory.errors import NodeWriteConflict

    if str(node.id) in store._nodes.get(scope.tenant_id, {}):
        raise NodeWriteConflict("memory identifier already exists")
    node.tenant_id = scope.tenant_id
    node.properties["scope_path"] = materialise_scope_path(scope)
    from symfonic.memory.backends.in_memory_owned_node_patch import new_incarnation
    new_incarnation(node)
    store._nodes[scope.tenant_id][str(node.id)] = node.model_copy(deep=True)
    return node

native_owner 

native_owner(path: Any) -> bool

Recognize canonical v1 owners without reinterpreting legacy keys.

Source code in src/symfonic/memory/backends/in_memory_ownership.py 
106
107
108
109
110
111
112
113
114
115
116
117
def native_owner(path: Any) -> bool:
    """Recognize canonical v1 owners without reinterpreting legacy keys."""
    if not isinstance(path, str) or "\x1f" in path:
        return False
    from symfonic.capabilities.memory.errors import MemoryContractError
    from symfonic.capabilities.memory.storage_key import StorageScopeKey

    try:
        StorageScopeKey.parse(path)
    except MemoryContractError:
        return False
    return True

preflight_privacy_v0 async 

preflight_privacy_v0(store: Any, subject: Any) -> None

Refuse owners a positional subject path cannot safely classify.

Source code in src/symfonic/memory/backends/in_memory_ownership.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
async def preflight_privacy_v0(store: Any, subject: Any) -> None:
    """Refuse owners a positional subject path cannot safely classify."""
    from symfonic.core.scope import TenantScope
    from symfonic.memory.backends.in_memory_transaction import writer

    scope = TenantScope.root("tenant", subject.tenant_id)
    if subject.child_segment is not None:
        scope = scope.child("sub_tenant", subject.child_segment)
    async with writer(store):
        for node in store._nodes.get(subject.tenant_id, {}).values():
            owner = stored_scope_path(node.properties, node.tenant_id)
            if _unproven_v0_owner(owner, scope):
                raise ValueError("legacy graph owner cannot prove subject privacy erasure")
        for edge in store._edges.get(subject.tenant_id, {}).values():
            owner = stored_scope_path(
                {"scope_path": getattr(edge, "_scope_path", None)}, edge.tenant_id,
            )
            if _unproven_v0_owner(owner, scope):
                raise ValueError("legacy graph edge cannot prove subject privacy erasure")

privacy_absent 

privacy_absent(store: Any, scope: Any) -> bool

Check nodes, owned edges, and workset cursors under the graph lock.

Source code in src/symfonic/memory/backends/in_memory_ownership.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def privacy_absent(store: Any, scope: Any) -> bool:
    """Check nodes, owned edges, and workset cursors under the graph lock."""
    from symfonic.memory.subtree import is_in_subtree

    root = materialise_scope_path(scope)
    for node in store._nodes.get(scope.tenant_id, {}).values():
        owner = stored_scope_path(node.properties, node.tenant_id, scope=scope)
        if _unproven_v0_owner(owner, scope) or (
            owner is not None and (_is_native_scope(scope) or not native_owner(owner))
            and is_in_subtree(owner, root)
        ):
            return False
    for edge in store._edges.get(scope.tenant_id, {}).values():
        owner = stored_scope_path(
            {"scope_path": getattr(edge, "_scope_path", None)}, edge.tenant_id, scope=scope,
        )
        if _unproven_v0_owner(owner, scope) or (
            owner is not None and (_is_native_scope(scope) or not native_owner(owner))
            and is_in_subtree(owner, root)
        ):
            return False
    return not any(is_in_subtree(key[0], root) for key in store._workset_cursors)

privacy_delete 

privacy_delete(store: Any, scope: Any) -> int

Sweep and permanently fence one storage subtree under the writer lock.

Source code in src/symfonic/memory/backends/in_memory_ownership.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def privacy_delete(store: Any, scope: Any) -> int:
    """Sweep and permanently fence one storage subtree under the writer lock."""
    from symfonic.core.scope import TenantScope
    from symfonic.memory.backends.in_memory_deletion import delete_subtree
    from symfonic.memory.storage_scope import NativeStorageScope
    from symfonic.memory.subtree import is_in_subtree

    if not isinstance(scope, (NativeStorageScope, TenantScope)):
        raise TypeError("privacy erase requires a storage scope")
    if getattr(store, "_write_owner", None) is not None:
        raise RuntimeError("privacy erase cannot join a rollback-capable graph transaction")
    erased = delete_subtree(store, scope)
    if not _is_native_scope(scope):
        root = scope.scope_path
        edges = store._edges.get(scope.tenant_id, {})
        for identifier, edge in tuple(edges.items()):
            owner = stored_scope_path(
                {"scope_path": getattr(edge, "_scope_path", None)},
                edge.tenant_id, scope=scope,
            )
            if not native_owner(owner) and is_in_subtree(owner, root):
                del edges[identifier]
    store._privacy_erased_subtrees[scope.tenant_id].add(scope.scope_path)
    if not _is_native_scope(scope) and scope.scope_path == legacy_scope_path(scope.tenant_id):
        store._privacy_erased_tenants.add(scope.tenant_id)
    return erased

read_edges 

read_edges(store: Any, scope: Any, *, subtree: bool = False) -> Any

Quarantine native edges with unproven owners or endpoints before LIMIT.

Source code in src/symfonic/memory/backends/in_memory_ownership.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def read_edges(store: Any, scope: Any, *, subtree: bool = False) -> Any:
    """Quarantine native edges with unproven owners or endpoints before LIMIT."""
    from symfonic.memory.pending import is_pending
    from symfonic.memory.retraction import is_retracted
    from symfonic.memory.subtree import is_in_subtree

    native = _is_native_scope(scope)
    prefixes = scope.ancestor_prefix_paths()
    root = materialise_scope_path(scope)

    def visible(properties: Any, tenant: str) -> bool:
        owner = stored_scope_path(properties, tenant, scope=scope)
        return owner is not None and (
            owner in prefixes or (subtree and is_in_subtree(owner, root))
        )

    for edge in store._edges.get(scope.tenant_id, {}).values():
        owner = getattr(edge, "_scope_path", None)
        if not native:
            if not native_owner(owner):
                yield edge
            continue
        if not visible({"scope_path": owner}, edge.tenant_id):
            continue
        endpoints = [store._nodes.get(scope.tenant_id, {}).get(str(identifier))
                     for identifier in (edge.source, edge.target)]
        if all(node is not None and visible(node.properties, node.tenant_id)
               and not is_pending(node.properties) and not is_retracted(node.properties)
               for node in endpoints):
            yield edge

refuse_privacy_erased 

refuse_privacy_erased(store: Any, scope: Any) -> None

Refuse v0 and v1 writes beneath a permanent privacy tombstone.

Source code in src/symfonic/memory/backends/in_memory_ownership.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def refuse_privacy_erased(store: Any, scope: Any) -> None:
    """Refuse v0 and v1 writes beneath a permanent privacy tombstone."""
    from symfonic.capabilities.memory.errors import MemoryContractError

    path = materialise_scope_path(scope)
    from symfonic.memory.subtree import is_in_subtree

    if scope.tenant_id in store._privacy_erased_tenants:
        raise MemoryContractError("native graph subject has been privacy-erased")
    if not _is_native_scope(scope) and store._privacy_erased_subtrees[scope.tenant_id]:
        root = legacy_scope_path(scope.tenant_id)
        if path == root or not is_in_subtree(path, root):
            raise MemoryContractError("native graph subject has been privacy-erased")
    if any(is_in_subtree(path, root) for root in store._privacy_erased_subtrees[scope.tenant_id]):
        raise MemoryContractError("native graph subject has been privacy-erased")

replace_node 

replace_node(store: Any, scope: Any, node: Any) -> Any

Existing unconditional add semantics, with a fresh row incarnation.

Source code in src/symfonic/memory/backends/in_memory_ownership.py 
145
146
147
148
149
150
151
152
153
def replace_node(store: Any, scope: Any, node: Any) -> Any:
    """Existing unconditional add semantics, with a fresh row incarnation."""
    check_node_owner(store, scope, node)
    node.tenant_id = scope.tenant_id
    node.properties["scope_path"] = materialise_scope_path(scope)
    from symfonic.memory.backends.in_memory_owned_node_patch import new_incarnation
    new_incarnation(node)
    store._nodes[scope.tenant_id][str(node.id)] = node.model_copy(deep=True)
    return node

serialized_write 

serialized_write(method: Any) -> Any

Check permanent privacy closure while holding the graph writer lock.

Source code in src/symfonic/memory/backends/in_memory_ownership.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
def serialized_write(method: Any) -> Any:
    """Check permanent privacy closure while holding the graph writer lock."""
    @wraps(method)
    async def locked(store: Any, *args: Any, **kwargs: Any) -> Any:
        from symfonic.memory.backends.in_memory_transaction import writer

        async with writer(store):
            if method.__name__ not in {
                "delete_node", "delete_edge", "delete_subtree", "delete_privacy_subtree",
            }:
                refuse_privacy_erased(store, args[0] if args else kwargs["scope"])
            return await method(store, *args, **kwargs)
    return locked