Skip to content

symfonic.memory.layers.procedural.dedup_audit

dedup_audit

Destructive-audit port for bulk procedural deduplication (CORE-178).

The memory capability cannot reach the platform audit seam, so it declares what it needs: an actor and a port that durably records intent before any mutation. A missing actor or port refuses without graph I/O. The refusal is an :class:AtomicDeduplicationUnavailable so every surface keeps its 503.

intent must raise when the record did not land. completed and failed describe what already happened; their failure is logged and never replaces the operation's own outcome.

DedupAuditContext dataclass 

DedupAuditContext(principal_id: str, caller: DedupCaller)

Who asked: an authenticated user or a configured service principal.

DedupAuditPort

Bases: Protocol

Records one bulk pass; intent raises unless durably recorded.

DedupAuditUnavailable

Bases: AtomicDeduplicationUnavailable

No actor or no working audit port; bulk dedup may not start.

require_dedup_audit 

require_dedup_audit(audit: DedupAuditPort | None, context: DedupAuditContext | None) -> tuple[DedupAuditPort, DedupAuditContext]

Refuse, without I/O, a pass that could not be attributed or recorded.

Source code in src/symfonic/memory/layers/procedural/dedup_audit.py 
72
73
74
75
76
77
78
79
80
81
82
def require_dedup_audit(
    audit: DedupAuditPort | None, context: DedupAuditContext | None,
) -> tuple[DedupAuditPort, DedupAuditContext]:
    """Refuse, without I/O, a pass that could not be attributed or recorded."""
    if audit is None:
        raise DedupAuditUnavailable("Bulk deduplication requires a configured audit sink")
    if not isinstance(context, DedupAuditContext):
        raise DedupAuditUnavailable(
            "Bulk deduplication requires an authenticated or configured principal"
        )
    return audit, context

run_audited_dedup async 

run_audited_dedup(audit: DedupAuditPort | None, context: DedupAuditContext | None, scope: TenantScope, operation: Callable[[], Awaitable[_T]]) -> _T

Await intent, run operation, then record its terminal outcome.

The scope key is derived from the exact scope being mutated, so a caller cannot attribute the pass to a different owner than the one it touches.

Source code in src/symfonic/memory/layers/procedural/dedup_audit.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
async def run_audited_dedup(
    audit: DedupAuditPort | None,
    context: DedupAuditContext | None,
    scope: TenantScope,
    operation: Callable[[], Awaitable[_T]],
) -> _T:
    """Await intent, run ``operation``, then record its terminal outcome.

    The scope key is derived from the exact scope being mutated, so a caller
    cannot attribute the pass to a different owner than the one it touches.
    """
    port, actor = require_dedup_audit(audit, context)
    scope_key = scope.scope_path
    await port.intent(actor, scope_key)
    try:
        result = await operation()
    except BaseException as exc:
        try:
            await port.failed(actor, scope_key, type(exc).__name__)
        except Exception:  # noqa: BLE001 - never mask the operation's error
            logger.exception("dedup failure audit did not land for %s", scope_key)
        raise
    try:
        await port.completed(actor, scope_key, int(result[0]))
    except Exception:  # noqa: BLE001 - the pass already committed
        logger.exception("dedup completion audit did not land for %s", scope_key)
    return result