Skip to content

symfonic.capabilities.memory.growth

growth

Gross new-edge quotas belong to a durable sweep, not a worker lifetime.

GrowthDeferred

Bases: BaseException

Named control outcome; legacy except Exception must not swallow it.

edge_or_defer async 

edge_or_defer(graph: Any, scope: Any, edge: Any, *, upsert: bool = False) -> Any

Single-edge atomic group: no invented edge on an explicitly deferred write.

Source code in src/symfonic/capabilities/memory/growth.py 
70
71
72
73
74
75
async def edge_or_defer(graph: Any, scope: Any, edge: Any, *, upsert: bool = False) -> Any:
    """Single-edge atomic group: no invented edge on an explicitly deferred write."""
    try:
        return await (graph.upsert_edge(scope, edge) if upsert else graph.add_edge(scope, edge))
    except GrowthDeferred:
        return None

reserve async 

reserve(overlay: Any, scope: Any, edge: Any, operation: str, limit: int) -> str | None

Reserve one genuinely new row before mutation; caller releases on failure.

Source code in src/symfonic/capabilities/memory/growth.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
async def reserve(overlay: Any, scope: Any, edge: Any, operation: str, limit: int) -> str | None:
    """Reserve one genuinely new row before mutation; caller releases on failure."""
    if not hasattr(overlay, "progress"):
        raise MemoryContractError("a growth budget requires durable bounded worksets")
    path = materialise_scope_path(scope)
    _, expected, _ = overlay.progress[path]
    if expected.upper is not None and expected.edge_limit != limit:
        raise MemoryContractError("cannot change the edge ceiling during a sweep")
    edges = overlay._edges[scope.tenant_id]  # noqa: SLF001 - owned journal projection
    existing = edges.get(str(edge.id))
    if existing is not None and (
        getattr(existing, "_scope_path", None) or legacy_scope_path(existing.tenant_id)
    ) != path:
        existing = None
    if operation == "upsert_edge":
        existing = next((row for row in edges.values() if
                         (row.source, row.target, row.relationship) ==
                         (edge.source, edge.target, edge.relationship)
                         and (getattr(row, "_scope_path", None)
                              or legacy_scope_path(row.tenant_id)) == path), existing)
    if existing is None:
        existing = await overlay.durable.workset_edge(
            scope, str(edge.source), str(edge.target), edge.relationship,
            identifier=str(edge.id) if operation == "add_edge" else None,
        )
        if existing is not None and any(
            (scope.tenant_id, str(n)) in overlay.cascaded_nodes
            for n in (existing.source, existing.target)
        ):
            existing = None
    if existing is not None and (scope.tenant_id, str(existing.id)) not in overlay.deleted_edges:
        return None
    used = expected.edges_created + overlay.growth_created.get(path, 0)
    if used >= limit:
        overlay.growth_deferred += 1
        raise GrowthDeferred("new-edge sweep budget exhausted")
    overlay.growth_created[path] = overlay.growth_created.get(path, 0) + 1
    return path