symfonic.core.learning.phases_procedural_promotion

phases_procedural_promotion

Phase 12: Promote episodic patterns to draft procedural skills.

Scans episodic memory for repeated action sequences. When a pattern appears min_pattern_count or more times across recent episodes, extracts it as a candidate procedural skill stored in draft status. Human reviewers approve/reject via /procedures/{id}/approve or /procedures/{id}/reject (shipped in Sprint 1).

Why always draft? Phase 12 is the offline learning path. It runs unattended during consolidation and should never auto-activate skills -- the real-time extraction path in AgentEngine._consolidate() is where MetacognitiveMiddleware gates live usage. Here, human review is the quality gate.

Pattern extraction is intentionally simple for v1:

• Entries whose metadata['action_type'] matches group by that key.
• Entries whose content starts with an action verb ("I ran", "I used", "Then I ") group by the verb + object phrase.
• (v7.4.7, opt-in via use_tool_calls_fallback) Entries whose metadata['tool_calls'] list is non-empty group by tool_calls[0]['name']. This lets the regex extractor produce signal for tool-call-heavy agents whose narrated content begins with "User: ..." (the engine's auto-turn summary shape) and therefore never matches the action-verb anchor.

If none of the heuristics fire, the episodic entry is skipped. A future ticket will upgrade this with an LLM-based extractor (see .claude/docs/2026-05-28-jarvio-llm-procedural-extractor-eval.md).

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/core/learning/phases_procedural_promotion.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 Jarvio 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:
        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