symfonic.core.learning.phases_profile

phases_profile

Phase 5: scoped profile-fact promotion (Deep Sleep plan, Stage 2).

Replaces the legacy apply_soul_corrections (formerly in phases.py), which was broken three ways at once:

1. It wrote instance VALUES into DomainTemplate.soul_schema -- a dict[str, str] mapping field names to their EXPECTED TYPES, consumed by extraction.txt as Schema: {{SOUL_SCHEMA}} to tell the model what shape a SOUL node should have. Writing values in place of types corrupted the schema shown to the model on the next turn.
2. Its result was discarded: the engine call site passed a defensive dict(domain.soul_schema) copy, mutated it in place, and never wrote it back -- report.soul_updates counted changes that were then garbage-collected.
3. It was a live cross-tenant accumulator: the scaffolder's worker template built the dict ONCE before the tenant loop and passed the SAME object to every tenant's consolidator.run(). In-place mutation meant tenant A's profile values could leak into the dict handed to tenant B.

promote_profile_corrections fixes all three structurally rather than by convention:

• It takes no caller-owned mutable dict in its signature at all -- there is nothing to accidentally share across tenants (kills bug 3).
• It is scope-keyed by construction: every read and write goes through GraphMemoryStore with the caller's TenantScope, so a promotion under tenant A's scope structurally cannot touch tenant B's data.
• It writes corrections onto the tenant's own SOUL: node(s) in the SEMANTIC layer -- never into a type schema. The type schema (profile_fields) is READ ONLY, to learn which field names count as "profile" (kills bug 1).
• The write IS the result (a graph mutation) -- there is no return value to discard (kills bug 2).

promote_profile_corrections async

promote_profile_corrections(
    graph: GraphMemoryStore,
    scope: TenantScope,
    recent_nodes: list[MemoryNode],
    profile_fields: frozenset[str],
) -> 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

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/core/learning/phases_profile.py

async def promote_profile_corrections(
    graph: GraphMemoryStore,
    scope: TenantScope,
    recent_nodes: list[MemoryNode],
    profile_fields: frozenset[str],
) -> 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).

    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("_last_edited_by") == "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:
        props = node.properties or {}
        for key in profile_fields:
            if key in props:
                corrected[key] = props[key]

    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)
    updates = 0
    for key, value in corrected.items():
        if existing.get(key) != value:
            merged[key] = value
            updates += 1

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

    merged["_last_edited_by"] = 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