Skip to content

symfonic.memory.janitor_bulk_plan

janitor_bulk_plan

Store-free characterization planner for procedural bulk deduplication.

The plan records observed preimages; applying it is not an atomic compare-and-swap. Public bulk entry points remain gated until a backend contract is proven.

BulkDedupAction dataclass 

BulkDedupAction(kind: str, node_id: str, expected_preimage: MemoryNode, canonical_update: MemoryNode | None = None)

One ordered legacy write, with the row seen before planning began.

BulkDedupPlan dataclass 

BulkDedupPlan(expected_preimages: dict[str, MemoryNode], canonical_updates: dict[str, MemoryNode], absorbed_ids: tuple[str, ...], merged_count: int, actions: tuple[BulkDedupAction, ...])

Decisions made entirely from detached, exact-owner procedural rows.

plan_bulk_dedup async 

plan_bulk_dedup(policy: _DedupPolicy, scope: TenantScope, rows: list[MemoryNode]) -> BulkDedupPlan

Replay the legacy greedy and review-authority decisions without store I/O.

Deep copies prevent merge implementations and callers from changing the store's live objects or the supplied snapshot during planning.

Source code in src/symfonic/memory/janitor_bulk_plan.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
async def plan_bulk_dedup(
    policy: _DedupPolicy, scope: TenantScope, rows: list[MemoryNode],
) -> BulkDedupPlan:
    """Replay the legacy greedy and review-authority decisions without store I/O.

    Deep copies prevent merge implementations and callers from changing the
    store's live objects or the supplied snapshot during planning.
    """
    nodes = [
        node.model_copy(deep=True) for node in rows
        if policy._is_exact_owner(scope, node)
    ]
    preimages = {str(node.id): node.model_copy(deep=True) for node in nodes}
    absorbed: set[str] = set()
    absorbed_order: list[str] = []
    processed: set[str] = set()
    updates: dict[str, MemoryNode] = {}
    actions: list[BulkDedupAction] = []

    def delete(node: MemoryNode) -> None:
        node_id = str(node.id)
        actions.append(BulkDedupAction("delete", node_id, preimages[node_id]))
        absorbed.add(node_id)
        absorbed_order.append(node_id)

    for i, primary in enumerate(nodes):
        if str(primary.id) in absorbed or str(primary.id) in processed:
            continue
        if policy._is_authored(primary):
            continue
        candidates = [
            node for j, node in enumerate(nodes)
            if j > i and str(node.id) not in absorbed
        ]
        duplicates = await policy.find_duplicates(scope, candidates, primary)
        cluster = [primary, *duplicates]
        reviewed = next(
            (
                candidate for candidate in cluster
                if any(
                    policy._reviewed_canonical_blocks_auto_draft(candidate, other)
                    for other in cluster if str(other.id) != str(candidate.id)
                )
            ),
            None,
        )
        if reviewed is not None:
            for draft in cluster:
                if (
                    str(draft.id) != str(reviewed.id)
                    and policy._reviewed_canonical_blocks_auto_draft(reviewed, draft)
                    and not policy._is_reviewed(draft)
                ):
                    delete(draft)
            primary = reviewed
            duplicates = [
                candidate for candidate in cluster
                if str(candidate.id) != str(reviewed.id)
                and str(candidate.id) not in absorbed
            ]
        for duplicate in duplicates:
            if policy._is_reviewed(primary) and policy._is_reviewed(duplicate):
                continue
            if policy._reviewed_canonical_blocks_auto_draft(primary, duplicate):
                delete(duplicate)
                continue
            if policy._reviewed_canonical_blocks_auto_draft(duplicate, primary):
                delete(primary)
                primary = duplicate
                continue
            if policy._is_reviewed(primary) or policy._is_reviewed(duplicate):
                continue
            if policy._automatic_terminal_and_draft_are_independent(primary, duplicate):
                continue
            if policy._automatic_and_manual_drafts_are_independent(primary, duplicate):
                continue
            canonical = await policy.merge(primary, duplicate)
            canonical = canonical.model_copy(deep=True)
            canonical_id = str(canonical.id)
            updates[canonical_id] = canonical
            actions.append(BulkDedupAction(
                "update", canonical_id, preimages[canonical_id], canonical,
            ))
            delete(duplicate)
            primary = canonical
        processed.add(str(primary.id))

    return BulkDedupPlan(
        preimages, updates, tuple(absorbed_order), len(absorbed_order), tuple(actions),
    )