Skip to content

symfonic.core.learning.promotion

promotion

v8.0 conversation→brand promotion + GDPR provenance (design §11/§12).

The 4-tier write model (§11.a) keeps conversation/working state ephemeral (transient durability) and promotes only extracted facts upward, out of band, during sleep consolidation. This module provides the PURE building blocks the SleepConsolidator adopts:

  • :func:is_promotion_eligible — the §11.c trigger (durable AND confidence ≥ floor; the consolidation-phase gate is the caller's responsibility since promotion only ever runs inside the consolidator).
  • :func:build_provenance / :func:merge_provenance — the §12.a provenance shape carried on promoted nodes (source_conversation_ids ARRAY etc.).
  • :func:apply_erasure — the §12.b reference-counted deletion cascade: remove a conversation id from the provenance; an empty array marks the node for deletion (default delete-on-empty-provenance PII policy), a surviving corroborating conversation keeps the fact.

These are deliberately side-effect-free so they are unit-testable in isolation and so wiring them into the consolidation hot path is a separate, reviewable integration step (the live SleepConsolidator is not rewired here).

apply_erasure 

apply_erasure(
    provenance: dict[str, Any] | None,
    erased_conversation_id: str,
    *,
    pii_policy: str = PII_POLICY_DELETE,
) -> tuple[dict[str, Any] | None, bool]

Apply a conversation-erasure to a promoted node's provenance (§12.b).

Returns (new_provenance, delete_node): - Remove erased_conversation_id from the source arrays. - If the array is now EMPTY: under the default delete-on-empty-provenance policy the node is a deletion candidate (delete_node=True) — no orphaned PII survives as a "brand asset". Under retain-anonymised the node survives with an empty provenance (adopter accepts responsibility). - If other conversations still corroborate the fact: it SURVIVES with the remaining provenance (reference-counted, NOT cascading-destructive).

Erasure operates on the SOURCE level (where it came from), never the query level (where it lives) — the §7 R3 mitigation, now locked.

Source code in src/symfonic/core/learning/promotion.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def apply_erasure(
    provenance: dict[str, Any] | None,
    erased_conversation_id: str,
    *,
    pii_policy: str = PII_POLICY_DELETE,
) -> tuple[dict[str, Any] | None, bool]:
    """Apply a conversation-erasure to a promoted node's provenance (§12.b).

    Returns ``(new_provenance, delete_node)``:
      - Remove ``erased_conversation_id`` from the source arrays.
      - If the array is now EMPTY: under the default
        ``delete-on-empty-provenance`` policy the node is a deletion
        candidate (``delete_node=True``) — no orphaned PII survives as a
        "brand asset".  Under ``retain-anonymised`` the node survives with an
        empty provenance (adopter accepts responsibility).
      - If other conversations still corroborate the fact: it SURVIVES with
        the remaining provenance (reference-counted, NOT cascading-destructive).

    Erasure operates on the SOURCE level (where it came from), never the
    query level (where it lives) — the §7 R3 mitigation, now locked.
    """
    if not provenance:
        # No provenance to erase from; nothing to do.  Not a deletion.
        return provenance, False
    conv_ids = [
        c for c in provenance.get("source_conversation_ids", [])
        if c != erased_conversation_id
    ]
    # Drop the matching scope_path by index alignment when present; otherwise
    # keep the remaining set (best-effort — the conversation ids are the
    # authoritative reference count).
    scope_paths = [
        sp for sp in provenance.get("source_scope_paths", [])
        if erased_conversation_id not in sp
    ]
    new_prov = {
        **provenance,
        "source_conversation_ids": conv_ids,
        "source_scope_paths": scope_paths,
    }
    if not conv_ids:
        if pii_policy == PII_POLICY_RETAIN:
            return new_prov, False
        # Default: delete the node — no orphaned PII.
        return new_prov, True
    return new_prov, False

build_provenance 

build_provenance(
    *,
    source_conversation_id: str,
    source_scope_path: str,
    promoted_by: str,
    extraction_confidence: float,
    promoted_at: str | None = None,
) -> dict[str, Any]

Build the §12.a provenance dict for a newly promoted node.

source_conversation_ids / source_scope_paths are ARRAYS from the start (a durable brand fact is frequently corroborated by multiple conversations over time; :func:merge_provenance appends to them).

Source code in src/symfonic/core/learning/promotion.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def build_provenance(
    *,
    source_conversation_id: str,
    source_scope_path: str,
    promoted_by: str,
    extraction_confidence: float,
    promoted_at: str | None = None,
) -> dict[str, Any]:
    """Build the §12.a provenance dict for a newly promoted node.

    ``source_conversation_ids`` / ``source_scope_paths`` are ARRAYS from the
    start (a durable brand fact is frequently corroborated by multiple
    conversations over time; :func:`merge_provenance` appends to them).
    """
    return {
        "source_conversation_ids": [source_conversation_id],
        "source_scope_paths": [source_scope_path],
        "promoted_at": promoted_at or datetime.now(UTC).isoformat(),
        "promoted_by": promoted_by,
        "extraction_confidence": extraction_confidence,
    }

extraction_confidence 

extraction_confidence(entry: Any) -> float

Read the extraction confidence off an entry's metadata.

Returns 0.0 when absent so an entry with no confidence signal never auto-promotes (conservative — design §11.c).

Source code in src/symfonic/core/learning/promotion.py 
43
44
45
46
47
48
49
50
51
52
53
54
def extraction_confidence(entry: Any) -> float:
    """Read the extraction confidence off an entry's metadata.

    Returns 0.0 when absent so an entry with no confidence signal never
    auto-promotes (conservative — design §11.c).
    """
    md = getattr(entry, "metadata", None) or {}
    raw = md.get("extraction_confidence", md.get("confidence"))
    try:
        return float(raw) if raw is not None else 0.0
    except (TypeError, ValueError):
        return 0.0

is_promotion_eligible 

is_promotion_eligible(
    entry: Any,
    *,
    confidence_floor: float = DEFAULT_PROMOTION_CONFIDENCE_FLOOR,
) -> bool

Return True when a conversation-tier fact may promote to brand (§11.c).

ALL must hold
  1. Durability gate — the source is durable (NOT transient/expired). Raw conversational state stays transient and never promotes; only extracted facts written durable are eligible.
  2. Confidence — extraction confidence ≥ confidence_floor.

The third condition (runs only in the out-of-band consolidation phase, §11.c.3) is structural: this predicate is ONLY called from the consolidator, so it is satisfied by construction. Org promotion (§11.c) is never automatic and is intentionally NOT covered here.

Source code in src/symfonic/core/learning/promotion.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def is_promotion_eligible(
    entry: Any,
    *,
    confidence_floor: float = DEFAULT_PROMOTION_CONFIDENCE_FLOOR,
) -> bool:
    """Return True when a conversation-tier fact may promote to brand (§11.c).

    ALL must hold:
      1. Durability gate — the source is ``durable`` (NOT transient/expired).
         Raw conversational state stays transient and never promotes; only
         extracted facts written ``durable`` are eligible.
      2. Confidence — extraction confidence ≥ ``confidence_floor``.

    The third condition (runs only in the out-of-band consolidation phase,
    §11.c.3) is structural: this predicate is ONLY called from the
    consolidator, so it is satisfied by construction.  Org promotion (§11.c)
    is never automatic and is intentionally NOT covered here.
    """
    if entry_durability(entry) != "durable":
        return False
    return extraction_confidence(entry) >= confidence_floor

merge_provenance 

merge_provenance(
    existing: dict[str, Any] | None,
    *,
    source_conversation_id: str,
    source_scope_path: str,
    promoted_by: str,
    extraction_confidence: float,
) -> dict[str, Any]

Append a corroborating conversation to an existing provenance (§12.a).

When the consolidation/janitor pipeline merges a duplicate fact, the reinforcing conversation's id is appended (de-duplicated). The highest observed confidence and the latest promoted_at win.

Source code in src/symfonic/core/learning/promotion.py 
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
def merge_provenance(
    existing: dict[str, Any] | None,
    *,
    source_conversation_id: str,
    source_scope_path: str,
    promoted_by: str,
    extraction_confidence: float,
) -> dict[str, Any]:
    """Append a corroborating conversation to an existing provenance (§12.a).

    When the consolidation/janitor pipeline merges a duplicate fact, the
    reinforcing conversation's id is appended (de-duplicated).  The highest
    observed confidence and the latest ``promoted_at`` win.
    """
    if not existing:
        return build_provenance(
            source_conversation_id=source_conversation_id,
            source_scope_path=source_scope_path,
            promoted_by=promoted_by,
            extraction_confidence=extraction_confidence,
        )
    conv_ids = list(existing.get("source_conversation_ids", []))
    scope_paths = list(existing.get("source_scope_paths", []))
    if source_conversation_id not in conv_ids:
        conv_ids.append(source_conversation_id)
    if source_scope_path not in scope_paths:
        scope_paths.append(source_scope_path)
    prior_conf = existing.get("extraction_confidence", 0.0)
    try:
        prior_conf = float(prior_conf)
    except (TypeError, ValueError):
        prior_conf = 0.0
    return {
        "source_conversation_ids": conv_ids,
        "source_scope_paths": scope_paths,
        "promoted_at": datetime.now(UTC).isoformat(),
        "promoted_by": promoted_by,
        "extraction_confidence": max(prior_conf, extraction_confidence),
    }