symfonic.core.learning.phases_procedural_promotion

phases_procedural_promotion

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

Phase 12 is on a roster the kernel composes, so it lives in the capability now and is imported back here for the legacy consolidator to call. One phase, one implementation, for as long as both routes ship.

promote_episodic_to_procedural async

promote_episodic_to_procedural(episodic_layer: Any, procedural_layer: ProceduralLayer, scope: TenantScope, *, min_pattern_count: int = DEFAULT_MIN_PATTERN_COUNT, recency_days: int = DEFAULT_RECENCY_DAYS, max_drafts_per_run: int = DEFAULT_MAX_DRAFTS_PER_RUN, use_tool_calls_fallback: bool = False, llm_extractor: Any | None = None, llm_model_name: str = '', promote_assistant_content: bool = False, gate: DurabilityGate | None = None) -> int

Phase 12: extract repeated episodic patterns into draft skills.

Algorithm

1. Fetch recent episodic entries (recency_days lookback).
2. Extract action patterns via _extract_pattern_key.
3. Keep patterns with count >= min_pattern_count.
4. Rank by frequency descending; cap at max_drafts_per_run.
5. For each, call procedural_layer.store_skill(..., status='draft'). SemanticMerge dedup inside store_skill handles near-duplicates against existing procedural nodes (draft or approved).
6. Return count of new draft nodes written this run.

Resilient to backends that don't populate created_at on episodic retrieval and to missing optional parameters (max_drafts_per_run=0 short-circuits).

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

async def promote_episodic_to_procedural(
    episodic_layer: Any,
    procedural_layer: ProceduralLayer,
    scope: TenantScope,
    *,
    min_pattern_count: int = DEFAULT_MIN_PATTERN_COUNT,
    recency_days: int = DEFAULT_RECENCY_DAYS,
    max_drafts_per_run: int = DEFAULT_MAX_DRAFTS_PER_RUN,
    use_tool_calls_fallback: bool = False,
    llm_extractor: Any | None = None,
    llm_model_name: str = "",
    promote_assistant_content: bool = False,
    gate: DurabilityGate | None = None,
) -> int:
    """Phase 12: extract repeated episodic patterns into draft skills.

    Algorithm:
      1. Fetch recent episodic entries (``recency_days`` lookback).
      2. Extract action patterns via ``_extract_pattern_key``.
      3. Keep patterns with ``count >= min_pattern_count``.
      4. Rank by frequency descending; cap at ``max_drafts_per_run``.
      5. For each, call ``procedural_layer.store_skill(..., status='draft')``.
         SemanticMerge dedup inside ``store_skill`` handles near-duplicates
         against existing procedural nodes (draft or approved).
      6. Return count of new draft nodes written this run.

    Resilient to backends that don't populate ``created_at`` on episodic
    retrieval and to missing optional parameters (``max_drafts_per_run=0``
    short-circuits).
    """
    if max_drafts_per_run <= 0:
        return 0

    try:
        entries = await episodic_layer.list_events(scope, limit=DEFAULT_TOP_K_EPISODIC)
    except Exception:
        logger.debug("Phase 12: episodic list_events failed", exc_info=True)
        return 0

    if not entries:
        return 0

    # Recency filter.
    cutoff = datetime.now(UTC) - timedelta(days=recency_days)
    recent: list[MemoryEntry] = []
    for e in entries:
        ts = _entry_created_at(e)
        if ts is None:
            # Unknown timestamp -> treat as recent (conservative: include).
            recent.append(e)
            continue
        if ts.tzinfo is None:
            ts = ts.replace(tzinfo=UTC)
        if ts >= cutoff:
            recent.append(e)

    if not recent:
        return 0

    # v7.6 adopter Ask 9 Option A: either-or dispatch.  When an LLM
    # extractor is supplied, run it INSTEAD of the regex extractor (the
    # eval rejected merged streams).  The shared draft-write path below
    # handles dedup + telemetry identically for both extractors.
    if llm_extractor is not None:
        from symfonic.capabilities.memory.phases.procedural_llm import (
            run_llm_extractor,
        )

        return await run_llm_extractor(
            llm_extractor,
            procedural_layer,
            scope,
            recent_episodes=recent,
            max_drafts_per_run=max_drafts_per_run,
            model_name=llm_model_name,
            promote_assistant_content=promote_assistant_content,
            gate=gate,
        )

    counts, contributors = _extract_action_patterns(
        recent,
        use_tool_calls_fallback=use_tool_calls_fallback,
        promote_assistant_content=promote_assistant_content,
        gate=gate,
    )
    if not counts:
        return 0

    # Rank patterns with >= min_pattern_count by frequency desc.
    eligible = [
        (name, cnt) for name, cnt in counts.most_common()
        if cnt >= min_pattern_count
    ]
    if not eligible:
        return 0

    # Snapshot of existing procedural content before we start -- used to
    # detect when SemanticMerge swallowed our write (returned canonical
    # existing node instead of creating a new one).
    try:
        existing_entries = await procedural_layer.query_skills(
            scope, "", top_k=1000, include_drafts=True,
        )
        existing_ids = {e.node_id for e in existing_entries if e.node_id}
    except Exception:
        existing_ids = set()

    created = 0
    # v7.4.8: count drafts that ``store_skill`` accepted but ``SemanticMerge``
    # collapsed into a pre-existing node.  Distinguishable from
    # ``store_skill`` exceptions (logged separately at L319) and from
    # patterns that never reach ``store_skill`` (filtered out by the
    # ``min_pattern_count`` gate above).  Surfaces in the summary log so
    # consumers can tell "found nothing" from "found patterns but they all
    # already existed" without instrumenting their own counters.
    drafts_rejected_by_dedup = 0
    for pattern_name, _cnt in eligible[:max_drafts_per_run]:
        skill_entry = _build_skill_entry(
            pattern_name, contributors[pattern_name], scope,
        )
        try:
            node = await procedural_layer.store_skill(
                scope, skill_entry, status="draft",
            )
        except Exception:
            logger.debug(
                "Phase 12: store_skill failed for %s", pattern_name,
                exc_info=True,
            )
            continue

        # If the janitor merged against a pre-existing node, don't count.
        if node.id in existing_ids:
            drafts_rejected_by_dedup += 1
            continue
        existing_ids.add(node.id)
        created += 1

    logger.info(
        "Phase 12: scanned=%d eligible_patterns=%d drafts_created=%d "
        "drafts_rejected_by_dedup=%d min_count=%d recency_days=%d",
        len(recent), len(eligible), created,
        drafts_rejected_by_dedup,
        min_pattern_count, recency_days,
    )
    return created