Rebuilding Aggregation Trees from EPCIS Events

When an auditor or a returns clerk asks a blunt question — what is currently packed inside pallet SSCC 007612345000012345 — the honest answer is not stored anywhere as a single row. It has to be rebuilt by replaying every AggregationEvent your EPCIS store ever captured for that hierarchy, in the exact order it happened, and folding the ADD and DELETE actions into a tree. This is the sharp edge of correct Parent-Child Serial Mapping: cached mapping tables drift and snapshots go stale, but the event log is append-only and, replayed correctly, it is authoritative. This guide builds that replay engine end to end, inside the broader discipline of Aggregation Hierarchy & Validation Workflows.

Prerequisites #

• Python 3.10+ — the fold and cycle-detection helpers below use X | Y unions and structural set operations.
• Pydantic v2 for a typed AggregationEvent contract, so a malformed record fails at the boundary instead of corrupting the in-memory tree.
• An EPCIS query client or export capable of returning AggregationEvent records for a given scope, ordered (or orderable) by eventTime.
• GS1 identifier literacy — child EPCs are SGTINs built from the same GTIN (01) and serial (21) that the unit’s commissioning ObjectEvent carries, alongside lot (10) and expiry (17); parent identifiers are SSCC-based EPC URIs for cases and pallets.
• Somewhere to persist the rebuilt tree — an in-memory cache, a graph table, or a materialized view — because replaying the full event history on every query does not scale past a modest event count.

Step-by-Step Solution #

Step 1 — Model the AggregationEvent contract #

Define the shape of a single record before touching the fold logic. Only two actions exist in GS1 aggregation semantics, and the parent must always be a container-class identifier.

from datetime import datetime, timezone
from typing import Literal
from pydantic import BaseModel, field_validator

class AggregationEvent(BaseModel):
    event_id: str
    event_time: datetime
    action: Literal["ADD", "DELETE"]
    parent_id: str            # SSCC EPC URI for the case or pallet
    child_epcs: list[str]     # SGTIN or nested SSCC EPC URIs

    @field_validator("event_time")
    @classmethod
    def _tz_aware(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("event_time must carry a timezone offset")
        return v.astimezone(timezone.utc)

    @field_validator("parent_id")
    @classmethod
    def _parent_is_container(cls, v: str) -> str:
        if "sscc" not in v:
            raise ValueError("parent_id must be an SSCC EPC URI for AggregationEvent")
        return v

DSCSA/GS1 note: AggregationEvent is the only EPCIS event type that expresses parent-child packaging relationships, and its action field admits exactly ADD or DELETE — there is no UPDATE, which is why replay order, not last-write-wins, is what determines the current tree.

Step 2 — Fetch and order events deterministically #

iterparse-style exports and paginated query APIs do not always guarantee stable ordering, especially when several events share a millisecond on a high-speed packaging line. Sort explicitly, with a secondary key, so two independent rebuild runs never diverge.

from operator import attrgetter

def ordered_events(raw_events: list[dict]) -> list[AggregationEvent]:
    """Parse and sort AggregationEvent records for deterministic replay."""
    events = [AggregationEvent.model_validate(e) for e in raw_events]
    # Primary key: event_time. Secondary key: event_id, so two events that
    # share a timestamp still replay in the same order on every rebuild.
    return sorted(events, key=attrgetter("event_time", "event_id"))

DSCSA/GS1 note: this mirrors the same ordering guarantee that keeps a decommission from ever being applied before its commission — here it keeps a DELETE from ever being folded before the ADD it is meant to undo.

Step 3 — Fold the stream into a parent-to-children map #

This is the core of the rebuild: walk the ordered events once, mutating a parent_id → set(child_epc) map. ADD inserts a child; DELETE removes it. A reverse index of each child’s current parent lets the fold catch a child that gets re-aggregated somewhere else without first being disaggregated — a violation the case-to-pallet aggregation validation checks are meant to prevent at commit time, but which a replay must still detect defensively when rebuilding from raw history.

def fold_aggregation_events(
    events: list[AggregationEvent],
) -> tuple[dict[str, set[str]], list[dict]]:
    """Fold ordered ADD/DELETE AggregationEvents into a parent -> children map.

    Returns (tree, anomalies). tree maps parent_id to its *current* child
    set. anomalies collects orphaned deletes and conflicting re-parenting
    so the caller can route them to a review queue instead of raising.
    """
    tree: dict[str, set[str]] = {}
    current_parent: dict[str, str] = {}   # child_epc -> its current parent_id
    anomalies: list[dict] = []

    for evt in events:
        children = tree.setdefault(evt.parent_id, set())

        if evt.action == "ADD":
            for child in evt.child_epcs:
                existing_parent = current_parent.get(child)
                if existing_parent and existing_parent != evt.parent_id:
                    # GS1 rule: a child must be disaggregated (DELETE)
                    # before it can be re-aggregated elsewhere.
                    anomalies.append({
                        "type": "reparent_without_delete",
                        "child": child,
                        "from_parent": existing_parent,
                        "to_parent": evt.parent_id,
                        "event_id": evt.event_id,
                    })
                    tree[existing_parent].discard(child)
                children.add(child)
                current_parent[child] = evt.parent_id

        elif evt.action == "DELETE":
            for child in evt.child_epcs:
                if child not in children:
                    anomalies.append({
                        "type": "orphan_delete",
                        "child": child,
                        "parent": evt.parent_id,
                        "event_id": evt.event_id,
                    })
                    continue
                children.discard(child)
                current_parent.pop(child, None)

    return tree, anomalies

DSCSA/GS1 note: an orphan_delete anomaly usually means the event history you replayed is incomplete — a DELETE arrived without its matching ADD in scope — which is exactly the drift pattern examined in fixing parent-child serial mapping drift post-decommission.

Step 4 — Guard against cycles before committing an ADD #

A tree cannot contain itself. If a case is accidentally re-aggregated onto a pallet that is already nested inside that same case — a realistic failure mode after a manual repackaging reaggregation correction — the fold must reject the edge rather than silently create a loop that later crashes any recursive “what’s inside” query.

def _descendants(tree: dict[str, set[str]], node: str) -> set[str]:
    """Collect every node currently nested, at any depth, under node."""
    seen: set[str] = set()
    stack = list(tree.get(node, ()))
    while stack:
        current = stack.pop()
        if current in seen:
            continue
        seen.add(current)
        stack.extend(tree.get(current, ()))
    return seen

def guard_against_cycles(
    tree: dict[str, set[str]], parent_id: str, child_epc: str
) -> bool:
    """Return True if aggregating child_epc under parent_id is safe.

    A cycle would mean parent_id is already nested somewhere beneath
    child_epc — for example a pallet accidentally re-aggregated onto a
    case that is itself already packed inside that same pallet.
    """
    return parent_id != child_epc and parent_id not in _descendants(tree, child_epc)

Wire this into Step 3’s ADD branch: call guard_against_cycles(tree, evt.parent_id, child) immediately before children.add(child), and if it returns False, append a cycle_detected anomaly instead of mutating the tree. This keeps the fold a pure, order-sensitive left-to-right pass with no separate cleanup phase.

DSCSA/GS1 note: cycle rejection protects the integrity of the aggregation hierarchy the same way a schema gate protects a raw ObjectEvent — a structurally impossible tree is a data-quality defect, not something to retry.

Step 5 — Answer “what is inside this SSCC right now” #

With the fold complete, resolving current contents is a plain recursive walk. It answers the original question at any depth, expanding nested cases inside a pallet down to the leaf units.

def contents_of(tree: dict[str, set[str]], sscc: str) -> dict[str, object]:
    """Answer 'what is inside this SSCC right now', expanded to full depth."""
    direct = sorted(tree.get(sscc, ()))
    return {
        "sscc": sscc,
        "direct_children": direct,
        "nested": [contents_of(tree, child) for child in direct if child in tree],
    }

DSCSA/GS1 note: this is the same query a trading partner or an inspector effectively asks during a suspect-product trace — the rebuilt tree must answer it without touching the raw event log again, which is why persisting tree after each fold matters operationally, not just for performance.

Verification #

Prove the fold is order-sensitive before trusting it against production history — a rebuild that ignores event order will silently produce a different, wrong tree.

from datetime import datetime, timezone

def _evt(event_id, action, parent, children, minute):
    return AggregationEvent(
        event_id=event_id,
        event_time=datetime(2026, 7, 1, 8, minute, tzinfo=timezone.utc),
        action=action,
        parent_id=parent,
        child_epcs=children,
    )

def test_fold_removes_deleted_child():
    pallet = "urn:epc:id:sscc:0312345.0000000111"
    case = "urn:epc:id:sscc:0312345.0000000222"
    unit = "urn:epc:id:sgtin:0312345.011111.SERIAL001"

    add_case = _evt("e1", "ADD", pallet, [case], 0)
    add_unit = _evt("e2", "ADD", case, [unit], 1)
    del_unit = _evt("e3", "DELETE", case, [unit], 2)

    tree, anomalies = fold_aggregation_events([add_case, add_unit, del_unit])
    assert tree[case] == set()          # unit was removed after being added
    assert tree[pallet] == {case}
    assert not anomalies

def test_out_of_order_delete_surfaces_as_orphan():
    case = "urn:epc:id:sscc:0312345.0000000222"
    unit = "urn:epc:id:sgtin:0312345.011111.SERIAL001"
    add_unit = _evt("e2", "ADD", case, [unit], 1)
    del_unit = _evt("e3", "DELETE", case, [unit], 2)

    # Replaying the DELETE before its matching ADD must surface an
    # anomaly, never a silent no-op — the failure mode an out-of-order
    # consumer produces.
    tree, anomalies = fold_aggregation_events([del_unit, add_unit])
    assert anomalies[0]["type"] == "orphan_delete"
    assert tree[case] == {unit}

Run both tests with pytest -q. In production, extend the second test into a scheduled reconciliation job: periodically re-fold the last N days of AggregationEvent history for a sampled set of SSCCs and diff the result against the cached tree — a nonzero diff means either the cache is stale or an anomaly slipped past ingestion-time validation.

Gotchas & Edge Cases #

• Millisecond-colliding timestamps. High-speed packaging lines can emit several AggregationEvent records within the same eventTime millisecond. Always sort on a stable secondary key such as event_id, or two rebuild runs over identical raw data can produce different trees.
• Late-arriving events after a network partition. An event that reaches the EPCIS store after you already cached a tree snapshot invalidates that snapshot. Either re-fold the affected parent’s full history or apply the late event as an incremental patch and re-run the cycle guard against the patched state.
• DELETE without a preceding ADD in scope. This usually means your query window or partial backfill excluded the original ADD, not that the data is truly orphaned — widen the replay window before treating it as a genuine anomaly.
• Re-parenting without disaggregation. Some legacy line software emits a bare ADD to a new parent when a case is manually moved, skipping the required DELETE from the old parent. The fold’s reverse index catches this, but the underlying process gap should be fixed upstream, not just tolerated in code.
• String, not integer, identifier comparisons. SSCCs and the GTIN component of an SGTIN can carry leading zeros; comparing or hashing them as integers anywhere in the tree silently merges distinct containers.

Related #

• Up to the parent topic: Parent-Child Serial Mapping
• Fixing Parent-Child Serial Mapping Drift Post-Decommission — repairing a tree once drift is already in production
• Handling Reaggregation After Repackaging — the upstream process that generates the ADD/DELETE pairs this replay consumes
• Aggregation Hierarchy & Validation Workflows — the domain this rebuild strategy belongs to