Skip to content

symfonic.capabilities.memory.promotion

promotion

Promotion: a memory moving toward the root, and the receipt for why.

Retrieval's visibility rule runs one way — a memory written at acme is visible inside acme/alice/s1, never the reverse — precisely so that publishing one session's content to a whole tenant is a decision somebody takes rather than a side effect of a query. This module is that decision.

Two rules, and both are refusals:

  • Only durable, corroborated facts promote. The 4-tier write model keeps conversational state transient and promotes extracted facts only; a confidence floor keeps a guess from becoming a tenant-wide truth. Same predicate as the shipped core.learning.promotion.is_promotion_eligible.
  • Promotion moves toward the root. Promoting downward would republish a tenant fact as one principal's own, which is a scope violation wearing a policy's clothes, so it raises rather than reporting.

Provenance is the other half. A promoted fact carries the conversations that produced it, as arrays from the start, because a durable fact is normally corroborated more than once — and because erasure is reference-counted: removing the last source conversation marks the memory for deletion, while a surviving corroboration keeps it. The dict shape is byte-identical to the shipped one (source_conversation_ids, source_scope_paths, promoted_at, promoted_by, extraction_confidence) so a node promoted by either side is readable by both.

Promotion dataclass 

Promotion(record: MemoryRecord, provenance: dict[str, Any], source_scope_path: str)

A promoted memory and the provenance that justifies it.

PromotionCandidate dataclass 

PromotionCandidate(record: MemoryRecord, confidence: float = 0.0, durability: str = _PROMOTABLE_DURABILITY, conversation_id: str = '')

One memory considered for promotion, with the signals that decide it.

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]

Remove one conversation from a provenance; report whether to delete.

Reference-counted rather than cascading: a fact corroborated by two conversations survives the erasure of one. When the last source goes, the default policy marks the memory for deletion — no orphaned personal data survives as an anonymous "fact".

Source code in src/symfonic/capabilities/memory/promotion.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def apply_erasure(
    provenance: dict[str, Any] | None,
    erased_conversation_id: str,
    *,
    pii_policy: str = PII_POLICY_DELETE,
) -> tuple[dict[str, Any] | None, bool]:
    """Remove one conversation from a provenance; report whether to delete.

    Reference-counted rather than cascading: a fact corroborated by two
    conversations survives the erasure of one. When the last source goes, the
    default policy marks the memory for deletion — no orphaned personal data
    survives as an anonymous "fact".
    """
    if not provenance:
        return provenance, False
    conversations = [
        conversation
        for conversation in provenance.get("source_conversation_ids", [])
        if conversation != erased_conversation_id
    ]
    scope_paths = [
        path
        for path in provenance.get("source_scope_paths", [])
        if erased_conversation_id not in path
    ]
    remaining = {
        **provenance,
        "source_conversation_ids": conversations,
        "source_scope_paths": scope_paths,
    }
    if conversations:
        return remaining, False
    return remaining, pii_policy != PII_POLICY_RETAIN

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]

The provenance a newly promoted memory carries.

Source code in src/symfonic/capabilities/memory/promotion.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
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]:
    """The provenance a newly promoted memory carries."""
    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,
    }

is_promotable 

is_promotable(candidate: PromotionCandidate, *, confidence_floor: float = DEFAULT_PROMOTION_CONFIDENCE_FLOOR) -> bool

Whether candidate may be published to a broader scope.

Source code in src/symfonic/capabilities/memory/promotion.py 
86
87
88
89
90
91
92
93
94
def is_promotable(
    candidate: PromotionCandidate,
    *,
    confidence_floor: float = DEFAULT_PROMOTION_CONFIDENCE_FLOOR,
) -> bool:
    """Whether ``candidate`` may be published to a broader scope."""
    if candidate.durability != _PROMOTABLE_DURABILITY:
        return False
    return candidate.confidence >= 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.

The highest confidence observed wins and the latest promotion time stands: a second conversation confirming a fact makes it more trustworthy, and taking the newer (possibly lower) confidence would let one weak restatement demote a well-established memory.

Source code in src/symfonic/capabilities/memory/promotion.py 
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
190
191
192
193
194
195
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.

    The highest confidence observed wins and the latest promotion time stands:
    a second conversation confirming a fact makes it *more* trustworthy, and
    taking the newer (possibly lower) confidence would let one weak restatement
    demote a well-established memory.
    """
    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,
        )
    conversations = list(existing.get("source_conversation_ids", []))
    scope_paths = list(existing.get("source_scope_paths", []))
    if source_conversation_id not in conversations:
        conversations.append(source_conversation_id)
    if source_scope_path not in scope_paths:
        scope_paths.append(source_scope_path)
    return {
        "source_conversation_ids": conversations,
        "source_scope_paths": scope_paths,
        "promoted_at": datetime.now(UTC).isoformat(),
        "promoted_by": promoted_by,
        "extraction_confidence": max(
            _as_float(existing.get("extraction_confidence")), extraction_confidence
        ),
    }

promote 

promote(candidate: PromotionCandidate, target: MemoryScope, *, promoted_by: str, promoted_at: str | None = None) -> Promotion

Restate candidate's memory at target, with its provenance.

The record id is kept. Ids are the upsert key within a scope, so re-promoting the same fact overwrites its own earlier promotion instead of accumulating near-duplicates of it, which is what makes running consolidation twice harmless.

Source code in src/symfonic/capabilities/memory/promotion.py 
 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
def promote(
    candidate: PromotionCandidate,
    target: MemoryScope,
    *,
    promoted_by: str,
    promoted_at: str | None = None,
) -> Promotion:
    """Restate ``candidate``'s memory at ``target``, with its provenance.

    The record *id is kept*. Ids are the upsert key within a scope, so
    re-promoting the same fact overwrites its own earlier promotion instead of
    accumulating near-duplicates of it, which is what makes running
    consolidation twice harmless.
    """
    source = candidate.record.scope
    if not target.covers(source):
        raise ScopeViolation(
            f"promotion from {source.path!r} to {target.path!r} moves a memory *away* "
            "from the root. Visibility runs one way: a memory at a broader scope is "
            "visible to every scope beneath it, so a downward promotion republishes "
            "one scope's content as another's."
        )
    record = MemoryRecord(
        record_id=candidate.record.record_id,
        layer=candidate.record.layer,
        text=candidate.record.text,
        scope_path=target.path,
        salience=candidate.record.salience,
        origin=f"promotion:{promoted_by}",
        revision=candidate.record.revision,
    )
    return Promotion(
        record=record,
        provenance=build_provenance(
            source_conversation_id=candidate.conversation_id,
            source_scope_path=source.path,
            promoted_by=promoted_by,
            extraction_confidence=candidate.confidence,
            promoted_at=promoted_at,
        ),
        source_scope_path=source.path,
    )