symfonic.core.learning.phases_profile

phases_profile

Phase 5, now owned by the memory capability.

promote_profile_corrections is on the quick roster, so it moved to :mod:symfonic.capabilities.memory.phases.profile and is imported back here for the legacy consolidator to call. One phase, one implementation, for as long as both routes ship -- see that module for the three bugs the phase was written to fix, which are worth keeping in front of anyone who edits it.

PROMOTED_FIELD_SOURCE module-attribute

PROMOTED_FIELD_SOURCE = 'user-correction'

Provenance recorded against a field this phase promoted.

The prompt block renders each profile field with its own provenance, so a role the user corrected today must not present the onboarding form's three-month-old stamp inherited from the node it landed on. Writing the field's real source here is what makes the rendered attribution true -- and provenance is recorded, never invented, so it is written at the moment of the write rather than reconstructed at render time.

promote_profile_corrections async

promote_profile_corrections(graph: GraphMemoryStore, scope: TenantScope, recent_nodes: list[MemoryNode], profile_fields: frozenset[str], now: datetime | None = None) -> int

Promote user-corrected profile facts onto the tenant's SOUL node.

Trigger condition (kept verbatim from the pre-redesign apply_soul_corrections, phases.py:446): only nodes carrying properties["_last_edited_by"] == "user_manual_edit" are treated as corrections. This keeps agent self-edits (extraction writes, this phase's own promotion writes, etc.) out of the promotion loop -- load-bearing, do not relax without also revisiting the idempotency guarantee below.

Parameters:

Name	Type	Description	Default
graph	GraphMemoryStore	Tenant-scoped graph store. Both the read (locating the canonical SOUL node) and the write (applying corrections) go through this store with scope, so a promotion is structurally confined to one tenant.	required
scope	TenantScope	Tenant isolation scope for every graph operation.	required
recent_nodes	list[MemoryNode]	Nodes considered for this consolidation pass (typically updated_at >= lookback cutoff). Any node satisfying the guard above is a candidate correction source, whether or not it is itself the canonical SOUL node.	required
profile_fields	frozenset[str]	The set of field names that constitute "profile" for this domain. Callers derive this from domain.soul_schema.keys() -- the schema is READ to learn which fields matter, but this function never writes to it. An empty set is a safe no-op (nothing is eligible to promote).	required
now	datetime | None	Instant recorded as each promoted field's provenance. Defaults to wall-clock UTC. Injectable because a caller that pins its own clock -- a test, a replay, a walkthrough -- would otherwise write a stamp it cannot predict, and the rendered provenance would read as being from the future.	None

Returns:

Type	Description
int	Count of profile fields whose value actually changed on the
int	canonical SOUL node (mirrors the legacy apply_soul_corrections
int	"count of key assignments made" semantics). A guard-marker-only
int	rewrite (see idempotency note below) does not increment this
int	count even though a graph write occurs.

Idempotency: when the correction source IS the canonical SOUL node (a human edited it directly rather than via a separate correction record), its own _last_edited_by guard is flipped from "user_manual_edit" to PROMOTION_MARKER even if the "corrected" values already match (i.e. even when the differential count is zero). Without this, the node would satisfy the guard again on the very next pass -- reading this promotion's own prior output back as a fresh user correction and re-promoting forever.

Source code in src/symfonic/capabilities/memory/phases/profile.py

async def promote_profile_corrections(
    graph: GraphMemoryStore,
    scope: TenantScope,
    recent_nodes: list[MemoryNode],
    profile_fields: frozenset[str],
    now: datetime | None = None,
) -> int:
    """Promote user-corrected profile facts onto the tenant's SOUL node.

    Trigger condition (kept verbatim from the pre-redesign
    ``apply_soul_corrections``, ``phases.py:446``): only nodes carrying
    ``properties["_last_edited_by"] == "user_manual_edit"`` are treated as
    corrections. This keeps agent self-edits (extraction writes, this
    phase's own promotion writes, etc.) out of the promotion loop --
    load-bearing, do not relax without also revisiting the idempotency
    guarantee below.

    Args:
        graph: Tenant-scoped graph store. Both the read (locating the
            canonical SOUL node) and the write (applying corrections) go
            through this store with ``scope``, so a promotion is
            structurally confined to one tenant.
        scope: Tenant isolation scope for every graph operation.
        recent_nodes: Nodes considered for this consolidation pass
            (typically ``updated_at >= lookback cutoff``). Any node
            satisfying the guard above is a candidate correction source,
            whether or not it is itself the canonical SOUL node.
        profile_fields: The set of field names that constitute "profile"
            for this domain. Callers derive this from
            ``domain.soul_schema.keys()`` -- the schema is READ to learn
            which fields matter, but this function never writes to it.
            An empty set is a safe no-op (nothing is eligible to promote).

        now: Instant recorded as each promoted field's provenance.
            Defaults to wall-clock UTC. Injectable because a caller that
            pins its own clock -- a test, a replay, a walkthrough -- would
            otherwise write a stamp it cannot predict, and the rendered
            provenance would read as being from the future.

    Returns:
        Count of profile fields whose value actually changed on the
        canonical SOUL node (mirrors the legacy ``apply_soul_corrections``
        "count of key assignments made" semantics). A guard-marker-only
        rewrite (see idempotency note below) does not increment this
        count even though a graph write occurs.

    Idempotency: when the correction source IS the canonical SOUL node
    (a human edited it directly rather than via a separate correction
    record), its own ``_last_edited_by`` guard is flipped from
    ``"user_manual_edit"`` to ``PROMOTION_MARKER`` even if the "corrected"
    values already match (i.e. even when the differential count is zero).
    Without this, the node would satisfy the guard again on the very next
    pass -- reading this promotion's own prior output back as a fresh
    user correction and re-promoting forever.
    """
    correction_nodes = [
        n for n in recent_nodes
        if (n.properties or {}).get(EDITED_BY_KEY) == "user_manual_edit"
    ]
    if not correction_nodes or not profile_fields:
        return 0

    # Last-writer-wins across every correction node touched this pass,
    # restricted to the caller-declared profile field set. Mirrors the
    # per-node accumulation order of the legacy function.
    corrected: dict[str, Any] = {}
    for node in correction_nodes:
        for key, value in _field_values(node).items():
            if key in profile_fields:
                corrected[key] = value

    if not corrected:
        return 0

    try:
        soul_nodes = await graph.query_nodes(
            scope, layer=MemoryLayer.SEMANTIC, label_prefix="SOUL",
        )
    except Exception:
        logger.debug(
            "promote_profile_corrections: query_nodes failed for tenant %s",
            scope.tenant_id, exc_info=True,
        )
        return 0

    target = _freshest(soul_nodes)
    if target is None:
        # No canonical profile node yet. The correction(s) stay flagged
        # user_manual_edit and are re-evaluated next pass -- harmless
        # (costs a query, not a mutation) and self-resolving once a SOUL
        # node is created (onboarding form / extractor).
        return 0

    existing = target.properties or {}
    merged = dict(existing)
    raw_provenance = existing.get("_field_provenance")
    provenance = dict(raw_provenance) if isinstance(raw_provenance, dict) else {}
    recorded_at = (now or datetime.now(UTC)).isoformat()
    updates = 0
    for key, value in corrected.items():
        if existing.get(key) != value:
            merged[key] = value
            # Per-field, because the node-level stamp answers "when was
            # this node written", which stops being the same question
            # once one node carries facts recorded at different times.
            provenance[key] = {
                "source": PROMOTED_FIELD_SOURCE,
                "recorded_at": recorded_at,
            }
            updates += 1
    if updates:
        merged["_field_provenance"] = provenance

    stale_guard = existing.get(EDITED_BY_KEY) == "user_manual_edit"
    if updates == 0 and not stale_guard:
        return 0

    merged[EDITED_BY_KEY] = PROMOTION_MARKER
    try:
        await graph.update_node(scope, target.id, {"properties": merged})
    except Exception:
        logger.debug(
            "promote_profile_corrections: update_node failed for %s (tenant %s)",
            target.id, scope.tenant_id, exc_info=True,
        )
        return 0

    return updates