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(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
| 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(*, 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
| 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(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
| 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(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
- Durability gate — the source is
durable (NOT transient/expired).
Raw conversational state stays transient and never promotes; only
extracted facts written durable are eligible.
- 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
| 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(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
| 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),
}
|