symfonic.core.learning.phases_maintenance

phases_maintenance

Moved to :mod:symfonic.capabilities.memory.phases.maintenance.

These phases are on a roster the kernel composes, so they live in the capability now and are imported back here for the legacy consolidator to call. One phase, one implementation, for as long as both routes ship.

The private helpers travel with them: the shipped per-phase tests import them by name, and a shim that re-exported only the public surface would have moved the code away from the tests that watch it.

cleanup_working_ttl async

cleanup_working_ttl(graph: GraphMemoryStore, scope: TenantScope, all_nodes: list[MemoryNode] | None = None) -> int

Prune WORKING-layer graph nodes with expired TTL.

Queries all WORKING-layer nodes and deletes those whose ttl_hours property has elapsed relative to updated_at.

v7.26.2 (Shape C1) adds two durability-aware branches:

• durability == "expired" -- pruned unconditionally (the explicit "this transient row served its purpose" signal from upgrade_durability), bypassing all TTL math.
• durability == "transient" -- subject to a 1h implicit TTL floor. A transient row with no ttl_hours ages out after 1h; a transient row with a longer ttl_hours is capped at 1h (Contract F). This protects against the "still syncing" footgun where a transient row with ttl_hours=720 would otherwise survive 30 days.

Nodes with no durability marker keep their pre-v7.26.2 behaviour exactly: pruned only when they carry a ttl_hours that has elapsed.

Unpublished writes are exempt. A persistent adapter spells "written but not yet flushed" as durability="transient" plus the namespaced markers in :mod:symfonic.memory.pending, because that is the only value the legacy reader already hides. It borrows the word, not the lifecycle: a pending row is resolved by a flush or a discard, never by a clock. Left unguarded, this phase would delete a turn's uncommitted writes an hour in and the flush that followed would commit nothing, silently.

Returns:

Type	Description
int	Number of pruned nodes.

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

async def cleanup_working_ttl(
    graph: GraphMemoryStore,
    scope: TenantScope,
    all_nodes: list[MemoryNode] | None = None,
) -> int:
    """Prune WORKING-layer graph nodes with expired TTL.

    Queries all WORKING-layer nodes and deletes those whose
    ``ttl_hours`` property has elapsed relative to ``updated_at``.

    v7.26.2 (Shape C1) adds two durability-aware branches:

    * ``durability == "expired"`` -- pruned unconditionally (the explicit
      "this transient row served its purpose" signal from
      ``upgrade_durability``), bypassing all TTL math.
    * ``durability == "transient"`` -- subject to a 1h implicit TTL floor.
      A transient row with no ``ttl_hours`` ages out after 1h; a transient
      row with a longer ``ttl_hours`` is capped at 1h (Contract F).  This
      protects against the "still syncing" footgun where a transient row
      with ``ttl_hours=720`` would otherwise survive 30 days.

    Nodes with no durability marker keep their pre-v7.26.2 behaviour
    exactly: pruned only when they carry a ``ttl_hours`` that has elapsed.

    **Unpublished writes are exempt.** A persistent adapter spells "written
    but not yet flushed" as ``durability="transient"`` plus the namespaced
    markers in :mod:`symfonic.memory.pending`, because that is the only value
    the legacy reader already hides. It borrows the word, not the lifecycle:
    a pending row is resolved by a flush or a discard, never by a clock. Left
    unguarded, this phase would delete a turn's uncommitted writes an hour in
    and the flush that followed would commit nothing, silently.

    Returns:
        Number of pruned nodes.
    """
    if all_nodes is None:
        working_nodes = await graph.query_nodes(scope, layer=MemoryLayer.WORKING)
    else:
        working_nodes = [
            n for n in all_nodes
            if n.layer == MemoryLayer.WORKING
        ]

    now = datetime.now(UTC)
    pruned = 0
    for node in working_nodes:
        props = node.properties or {}
        if is_pending(props):
            continue  # Resolved by flush or discard, never by a clock.
        durability = props.get("durability", "durable")

        # v7.26.2: explicit "expired" marker prunes regardless of TTL.
        if durability == "expired":
            try:
                await graph.delete_node(scope, node.id)
                pruned += 1
                logger.info(
                    "Durability=expired, pruned working node: %s", node.id,
                )
            except Exception:
                logger.debug(
                    "Failed to prune expired node %s", node.id, exc_info=True,
                )
            continue

        ttl_raw = props.get("ttl_hours")
        if ttl_raw is None:
            # v7.26.2: transient rows without an explicit TTL get a 1h floor.
            if durability == "transient":
                ttl_hours = 1.0
            else:
                continue
        else:
            try:
                ttl_hours = float(str(ttl_raw))
            except (ValueError, TypeError):
                continue
            # v7.26.2: transient rows are capped at the 1h floor (Contract F).
            if durability == "transient":
                ttl_hours = min(ttl_hours, 1.0)

        if node.updated_at is None:
            continue
        expiry = node.updated_at + timedelta(hours=ttl_hours)
        if now >= expiry:
            try:
                await graph.delete_node(scope, node.id)
                pruned += 1
                logger.info("Working TTL expired, pruned node: %s", node.id)
            except Exception:
                logger.debug(
                    "Failed to prune working TTL node %s", node.id, exc_info=True,
                )
    return pruned

decay_importance async

decay_importance(graph: GraphMemoryStore, scope: TenantScope, all_nodes: list[MemoryNode], stale_days: int = STALE_DAYS_DEFAULT) -> int

Reduce importance of semantic nodes not accessed in stale_days.

Nodes whose updated_at is older than the threshold have their importance reduced by 0.5, floored at 1.0.

High-signal / identity nodes are EXEMPT from decay so that the user profile and agent persona are never corrupted by age-based attrition. See :func:_is_exempt_from_decay for the exemption criteria.

Returns:

Type	Description
int	Number of decayed nodes.

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

async def decay_importance(
    graph: GraphMemoryStore,
    scope: TenantScope,
    all_nodes: list[MemoryNode],
    stale_days: int = STALE_DAYS_DEFAULT,
) -> int:
    """Reduce importance of semantic nodes not accessed in ``stale_days``.

    Nodes whose ``updated_at`` is older than the threshold have their
    importance reduced by 0.5, floored at 1.0.

    High-signal / identity nodes are EXEMPT from decay so that the user
    profile and agent persona are never corrupted by age-based attrition.
    See :func:`_is_exempt_from_decay` for the exemption criteria.

    Returns:
        Number of decayed nodes.
    """
    cutoff = datetime.now(UTC) - timedelta(days=stale_days)
    decayed = 0
    for node in all_nodes:
        if node.updated_at is None or node.updated_at >= cutoff:
            continue
        if _is_exempt_from_decay(node):
            logger.debug(
                "decay_importance: exempted %s (importance=%.1f, label=%s)",
                node.id, node.importance, node.label,
            )
            continue
        new_importance = max(node.importance - DECAY_AMOUNT, MIN_IMPORTANCE)
        if new_importance == node.importance:
            continue
        try:
            await graph.update_node(scope, node.id, {"importance": new_importance})
            decayed += 1
            logger.info(
                "Importance decayed: node=%s %.1f -> %.1f",
                node.id, node.importance, new_importance,
            )
        except Exception:
            logger.debug("Failed to decay node %s", node.id, exc_info=True)
    return decayed

is_identity_labelled

is_identity_labelled(node: MemoryNode) -> bool

Return True when node's label carries an identity prefix.

Label starts with SOUL: / SOUL (user profile data) or AGENT_IDENTITY: / AGENT_IDENTITY (agent persona). Split out of :func:_is_exempt_from_decay (importance-decay's exemption bundles this with the importance >= CRITICAL_IMPORTANCE_THRESHOLD check) so :mod:phases_semantic_expiry can reuse the label-only half without also inheriting the importance-based exemption -- see that module's docstring for why a high-importance node must NOT be exempt there.

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

def is_identity_labelled(node: MemoryNode) -> bool:
    """Return True when ``node``'s label carries an identity prefix.

    Label starts with ``SOUL:`` / ``SOUL `` (user profile data) or
    ``AGENT_IDENTITY:`` / ``AGENT_IDENTITY `` (agent persona). Split out
    of :func:`_is_exempt_from_decay` (importance-decay's exemption bundles
    this with the ``importance >= CRITICAL_IMPORTANCE_THRESHOLD`` check)
    so :mod:`phases_semantic_expiry` can reuse the label-only half without
    also inheriting the importance-based exemption -- see that module's
    docstring for why a high-importance node must NOT be exempt there.
    """
    label = node.label or ""
    return any(label.startswith(prefix) for prefix in _EXEMPT_LABEL_PREFIXES)

prune_retracted async

prune_retracted(graph: GraphMemoryStore, scope: TenantScope, grace_days: int = RETRACTED_GRACE_DAYS_DEFAULT) -> int

Hard-delete nodes soft-retracted more than grace_days ago.

A retract_node extraction op flags a false-positive/corrected memory with the namespaced retraction marker + timestamp (see symfonic.memory.retraction and SymfonicAgent._retract_memory). The flag alone removes the node from every read path immediately (GraphMemoryStore.query_nodes excludes it), but the row lingers for reversibility/audit. This phase reclaims it once it has been retracted past the grace window.

Must query with include_retracted=True -- retracted nodes are invisible to the default read path by design, so a plain scan would never see them.

query_nodes is ancestor-visible (a child scope sees a parent scope's rows), so an enumerated retracted node may belong to an ANCESTOR scope, not this one. Only a node materialized at exactly scope is deleted; ancestor-owned rows are left for the ancestor's own consolidation to reclaim, otherwise a child's consolidation could hard-delete a parent's memory.

Returns:

Type	Description
int	Number of pruned nodes.

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

async def prune_retracted(
    graph: GraphMemoryStore,
    scope: TenantScope,
    grace_days: int = RETRACTED_GRACE_DAYS_DEFAULT,
) -> int:
    """Hard-delete nodes soft-retracted more than ``grace_days`` ago.

    A ``retract_node`` extraction op flags a false-positive/corrected memory
    with the namespaced retraction marker + timestamp (see
    ``symfonic.memory.retraction`` and ``SymfonicAgent._retract_memory``). The
    flag alone removes the node from every read path immediately
    (``GraphMemoryStore.query_nodes`` excludes it), but the row lingers for
    reversibility/audit. This phase reclaims it once it has been retracted
    past the grace window.

    Must query with ``include_retracted=True`` -- retracted nodes are invisible
    to the default read path by design, so a plain scan would never see them.

    ``query_nodes`` is ancestor-visible (a child scope sees a parent scope's
    rows), so an enumerated retracted node may belong to an ANCESTOR scope,
    not this one. Only a node materialized at exactly ``scope`` is deleted;
    ancestor-owned rows are left for the ancestor's own consolidation to
    reclaim, otherwise a child's consolidation could hard-delete a parent's
    memory.

    Returns:
        Number of pruned nodes.
    """
    try:
        nodes = await graph.query_nodes(scope, include_retracted=True)
    except TypeError:
        # Store without the include_retracted kwarg (older/custom backend):
        # nothing to reclaim through this path.
        return 0

    cutoff = datetime.now(UTC) - timedelta(days=grace_days)
    write_scope = materialise_scope_path(scope)
    pruned = 0
    for node in nodes:
        props = node.properties or {}
        if not is_retracted(props):
            continue
        if stored_scope_path(props, node.tenant_id) != write_scope:
            continue
        # Within the grace window -> keep (reversible). Missing/unparseable
        # timestamp -> KEEP (never reclaim): a marker without a trustworthy
        # timestamp cannot be proven to be past the grace window, and treating
        # it as long-retracted would let arbitrary domain data that happens to
        # carry the marker (or a malformed stamp) be silently hard-deleted.
        retracted_at = props.get(RETRACTED_AT_KEY)
        if retracted_at is None:
            continue
        try:
            ts = datetime.fromisoformat(str(retracted_at))
        except (ValueError, TypeError):
            continue
        if ts.tzinfo is None:
            ts = ts.replace(tzinfo=UTC)
        if ts >= cutoff:
            continue
        try:
            await graph.delete_node(scope, node.id)
            pruned += 1
            logger.info(
                "Pruned retracted node: %s (reason=%r)",
                node.id, props.get(RETRACTION_REASON_KEY),
            )
        except Exception:
            logger.debug(
                "Failed to prune retracted node %s", node.id, exc_info=True,
            )
    return pruned