Skip to content

symfonic.platform.edge_curation

edge_curation

Verified, pre-audited edge deletion over the native graph service.

EdgeCurationService 

EdgeCurationService(edges, *, sink=None)
Source code in src/symfonic/platform/edge_curation.py 
13
14
15
def __init__(self, edges, *, sink=None):
    self._edges = edges
    self._sink = sink

delete_many async 

delete_many(principal, edge_ids)

Audit every intent first, then retain the legacy attempted-call count.

Source code in src/symfonic/platform/edge_curation.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
async def delete_many(self, principal, edge_ids):
    """Audit every intent first, then retain the legacy attempted-call count."""
    self._edges.ensure_available()
    if (self._sink is None or isinstance(self._sink, NullAuditSink)
            or not callable(getattr(self._sink, "emit", None))):
        raise AuditSinkError("edge deletion requires an explicitly configured audit sink")
    audit = AuditSeam(self._sink)

    def event(edge_id, outcome):
        return AuditRecord(
            action="bulk_delete_edge", outcome=outcome,
            principal_id=principal.principal_id, scope_key=principal.scope.scope_key,
            resource_type="memory_edge", resource_id=edge_id,
            metadata={"tenant_id": principal.scope.tenant_id},
        )

    for edge_id in edge_ids:
        await audit.record(event(edge_id, "initiated"), destructive=True)
    deleted = 0
    for edge_id in edge_ids:
        try:
            await self._edges.delete(memory_scope(principal.scope), edge_id)
        except Exception:  # noqa: BLE001 - preserve legacy per-ID continuation
            logger.warning("Bulk edge deletion failed for one ID")
            await audit.record(event(edge_id, "failed"))
        else:
            deleted += 1
            await audit.record(event(edge_id, "completed"))
    return {"deleted": deleted}