Implement Peer Learning Goals and Stage Fit Enhancements

- Introduced `_peer_stage_learning_goals` to retrieve learning goals from peer stages, enhancing the ability to filter exercises based on cross-slot collisions.
- Added `_filter_learning_goal_candidate_ids` to refine candidate selection by incorporating peer learning goals and stage fit criteria, improving exercise relevance in suggestions.
- Enhanced `pick_best_path_hit` and `_match_roadmap_slot` to utilize peer learning goals for better exercise selection and to prevent conflicts with titles from other stages.
- Updated `stage_refinement_criteria_from_learning_goal` to provide clearer criteria for stage refinement based on learning goals.
- Bumped version to 0.8.229 to reflect the new features and improvements.

backend/planning_exercise_path_builder.py

@@ -53,6 +53,8 @@ from planning_exercise_semantics import (
     resolve_path_anti_patterns,
     resolve_path_primary_topic,
     exercise_passes_path_semantic_gate,
+    exercise_passes_stage_fit,
+    exercise_title_matches_peer_stage_goal,
     pick_best_path_hit,
     resolve_semantic_skill_weights,
     step_phase_for_index,
@@ -201,6 +203,78 @@ def _roadmap_structured_from_body(body: ProgressionPathSuggestRequest) -> Option
     )
 
 
+def _peer_stage_learning_goals(
+    roadmap_ctx: ProgressionRoadmapContext,
+    *,
+    current_major_index: int,
+) -> List[str]:
+    goals: List[str] = []
+    for spec in roadmap_ctx.stage_specs or []:
+        if int(spec.major_step_index) == int(current_major_index):
+            continue
+        lg = (spec.learning_goal or "").strip()
+        if lg and lg not in goals:
+            goals.append(lg)
+    return goals
+
+
+def _filter_learning_goal_candidate_ids(
+    cur,
+    *,
+    tenant: TenantContext,
+    progression_graph_id: Optional[int],
+    candidate_ids: Sequence[int],
+    stage_goal: str,
+    stage_match_brief: PlanningSemanticBrief,
+    stage_anti: Optional[List[str]],
+    path_primary: str,
+    path_tech_excludes: Optional[List[str]],
+    peer_learning_goals: Sequence[str],
+) -> List[int]:
+    """Learning-Goal-Kandidaten nur, wenn sie Stufen-Gate und Peer-Check bestehen."""
+    if not candidate_ids:
+        return []
+    vis_sql, vis_params = _planning_visibility_sql(cur, tenant, progression_graph_id)
+    rows = _load_supplemental_exercise_rows(
+        cur,
+        tenant=tenant,
+        progression_graph_id=progression_graph_id,
+        exercise_ids=list(candidate_ids),
+        vis_sql=vis_sql,
+        vis_params=vis_params,
+    )
+    out: List[int] = []
+    for row in rows:
+        try:
+            eid = int(row.get("id") or 0)
+        except (TypeError, ValueError):
+            continue
+        if eid <= 0:
+            continue
+        title = str(row.get("title") or "")
+        if peer_learning_goals and exercise_title_matches_peer_stage_goal(
+            title,
+            current_learning_goal=stage_goal,
+            peer_learning_goals=peer_learning_goals,
+        ):
+            continue
+        summary = str(row.get("summary") or "")
+        goal_text = str(row.get("goal") or row.get("exercise_goal") or "")
+        if exercise_passes_stage_fit(
+            learning_goal=stage_goal,
+            title=title,
+            summary=summary,
+            goal=goal_text,
+            stage_brief=stage_match_brief,
+            anti_patterns=stage_anti,
+            path_primary_topic=path_primary or None,
+            path_technique_excludes=path_tech_excludes,
+            relaxed=True,
+        ):
+            out.append(eid)
+    return out
+
+
 def _pick_best_path_hit(
     hits: List[Dict[str, Any]],
     used_exercise_ids: Set[int],
@@ -212,6 +286,7 @@ def _pick_best_path_hit(
     stage_match_brief: Optional[PlanningSemanticBrief] = None,
     path_primary_topic: Optional[str] = None,
     path_technique_excludes: Optional[List[str]] = None,
+    peer_learning_goals: Optional[List[str]] = None,
 ) -> Optional[Dict[str, Any]]:
     return pick_best_path_hit(
         hits,
@@ -223,6 +298,7 @@ def _pick_best_path_hit(
         stage_match_brief=stage_match_brief,
         path_primary_topic=path_primary_topic,
         path_technique_excludes=path_technique_excludes,
+        peer_learning_goals=peer_learning_goals,
     )
 
 
@@ -366,13 +442,12 @@ def _fetch_learning_goal_library_candidate_ids(
     learning_goal: str,
     limit: int = 24,
 ) -> List[int]:
-    """Sichtbare Übungen, deren Titel/Volltext zum Stufen-Lernziel passt."""
+    """Sichtbare Übungen mit exakt passendem Titel oder Volltext-Treffer (kein breites LIKE)."""
     lg = (learning_goal or "").strip()
     if len(lg) < 3:
         return []
     vis_sql, vis_params = _planning_visibility_sql(cur, tenant, progression_graph_id)
     tsq = _safe_tsquery_fragment(lg)
-    like_pat = f"%{lg[:100].lower()}%"
     try:
         cur.execute(
             f"""
@@ -382,7 +457,6 @@ def _fetch_learning_goal_library_candidate_ids(
               AND COALESCE(e.status, '') <> %s
               AND (
                 lower(trim(e.title)) = lower(trim(%s))
-                OR lower(e.title) LIKE %s
                 OR (%s <> '' AND e.search_vector @@ plainto_tsquery('german', %s))
               )
             ORDER BY
@@ -395,7 +469,6 @@ def _fetch_learning_goal_library_candidate_ids(
                 *vis_params,
                 "archived",
                 lg,
-                like_pat,
                 tsq,
                 tsq,
                 lg,
@@ -411,14 +484,11 @@ def _fetch_learning_goal_library_candidate_ids(
             FROM exercises e
             WHERE ({vis_sql})
               AND COALESCE(e.status, '') <> %s
-              AND (
+              AND lower(trim(e.title)) = lower(trim(%s))
-                lower(trim(e.title)) = lower(trim(%s))
+            ORDER BY e.id ASC
-                OR lower(e.title) LIKE %s
-              )
-            ORDER BY CASE WHEN lower(trim(e.title)) = lower(trim(%s)) THEN 0 ELSE 1 END, e.id ASC
             LIMIT %s
             """,
-            [*vis_params, "archived", lg, like_pat, lg, int(limit)],
+            [*vis_params, "archived", lg, int(limit)],
         )
     out: List[int] = []
     for row in cur.fetchall() or []:
@@ -1092,14 +1162,30 @@ def _match_roadmap_slot(
         major_step=major,
     )
     step_kind = resolve_step_exercise_kind_filter(stage_spec, body.exercise_kind_any)
+    peer_goals = _peer_stage_learning_goals(
+        roadmap_ctx,
+        current_major_index=int(stage_spec.major_step_index),
+    )
 
     supplemental_ids = _supplemental_exercise_ids_from_body(cur, body)
-    lg_candidates = _fetch_learning_goal_library_candidate_ids(
+    lg_candidates_raw = _fetch_learning_goal_library_candidate_ids(
         cur,
         tenant=tenant,
         progression_graph_id=body.progression_graph_id,
         learning_goal=stage_goal,
     )
+    lg_candidates = _filter_learning_goal_candidate_ids(
+        cur,
+        tenant=tenant,
+        progression_graph_id=body.progression_graph_id,
+        candidate_ids=lg_candidates_raw,
+        stage_goal=stage_goal,
+        stage_match_brief=stage_match_brief,
+        stage_anti=stage_anti,
+        path_primary=path_primary,
+        path_tech_excludes=path_tech_excludes,
+        peer_learning_goals=peer_goals,
+    )
     supplemental_ids = list(
         dict.fromkeys(
             int(x)
@@ -1163,6 +1249,7 @@ def _match_roadmap_slot(
         stage_match_brief=stage_match_brief,
         path_primary_topic=path_primary or None,
         path_technique_excludes=path_tech_excludes or None,
+        peer_learning_goals=peer_goals,
     )
 
     if not hit:

backend/planning_exercise_semantics.py

@@ -864,19 +864,48 @@ def stage_focus_phrases_from_learning_goal(learning_goal: str) -> List[str]:
         return []
     tokens = _significant_stage_tokens(lg, strip_negated=True)
     phrases: List[str] = []
-    for tok in tokens:
+    norm_lg = _normalize_phrase(lg)
-        if len(tok) >= 5 and tok not in phrases:
+    if len(norm_lg) >= 8:
-            phrases.append(tok)
+        phrases.append(norm_lg[:120])
     for i in range(len(tokens) - 1):
         pair = f"{tokens[i]} {tokens[i + 1]}"
         if len(pair) >= 8 and pair not in phrases:
             phrases.append(pair)
-    norm_lg = _normalize_phrase(lg)
+    for tok in tokens:
-    if len(norm_lg) >= 8 and norm_lg not in phrases:
+        if len(tok) >= 6 and tok not in phrases:
-        phrases.insert(0, norm_lg[:120])
+            phrases.append(tok)
     return phrases[:8]
 
 
+def stage_refinement_criteria_from_learning_goal(learning_goal: str) -> List[str]:
+    """Erfolgskriterien für Phase C — nur aussagekräftige Mehrwort-Phrasen."""
+    out: List[str] = []
+    for phrase in stage_focus_phrases_from_learning_goal(learning_goal):
+        p = str(phrase or "").strip()
+        if not p:
+            continue
+        if " " in p or len(p) >= 12:
+            out.append(p[:120])
+    return out[:4]
+
+
+def exercise_title_matches_peer_stage_goal(
+    title: str,
+    *,
+    current_learning_goal: str,
+    peer_learning_goals: Sequence[str],
+) -> bool:
+    """Titel passt zum Lernziel einer anderen Roadmap-Stufe (Cross-Slot-Kollision)."""
+    current = (current_learning_goal or "").strip()
+    for peer in peer_learning_goals or []:
+        plg = (peer or "").strip()
+        if len(plg) < 3 or plg == current:
+            continue
+        if exercise_title_equivalent_to_stage_goal(title, plg):
+            return True
+    return False
+
+
 def _significant_stage_tokens(learning_goal: str, *, strip_negated: bool = True) -> List[str]:
     """Wörter aus Stufen-Lernziel für Text-Match (ohne Füllwörter, ohne Negationssegmente)."""
     text = _normalize_phrase(learning_goal)
@@ -1356,17 +1385,23 @@ def _pick_roadmap_rank_fallback(
     stage_anti_patterns: Optional[Sequence[str]] = None,
     path_primary_topic: Optional[str] = None,
     path_technique_excludes: Optional[Sequence[str]] = None,
+    stage_match_brief: Optional[PlanningSemanticBrief] = None,
+    peer_learning_goals: Optional[Sequence[str]] = None,
 ) -> Optional[Dict[str, Any]]:
     """
     Roadmap-Notfall: bester Treffer nach Stufen-Ranking, wenn striktes Gate leer läuft.
 
-    Filtert weiterhin Ausschlüsse und Technik-Scope (Kumite etc.), aber ohne
+    Weiterhin mit relaxed stage_fit — kein blindes Ranking ohne Stufen-Passung.
-    Mindest-Semantik-Schwelle — so finden auch wortnahe Bibliotheks-Übungen den Slot.
     """
     stage_goal = (stage_learning_goal or "").strip()
     if not stage_goal or not hits:
         return None
 
+    stage_brief = stage_match_brief or build_stage_match_brief(
+        learning_goal=stage_goal,
+        anti_patterns=stage_anti_patterns,
+    )
+
     best: Optional[Dict[str, Any]] = None
     best_key: Tuple[float, float] = (-1.0, -1.0)
     for hit in hits:
@@ -1379,33 +1414,31 @@ def _pick_roadmap_rank_fallback(
         title = str(hit.get("title") or "")
         summary = str(hit.get("summary") or "")
         goal_text = str(hit.get("goal") or hit.get("exercise_goal") or "")
-        blob = _blob_from_fields(title, summary, goal_text, [])
+        if peer_learning_goals and exercise_title_matches_peer_stage_goal(
-        exclude_phrases = merge_stage_exclude_phrases(stage_goal, stage_anti_patterns)
+            title,
-        if exclude_phrases and _blob_matches_stage_excludes(blob, exclude_phrases):
+            current_learning_goal=stage_goal,
+            peer_learning_goals=peer_learning_goals,
+        ):
             continue
-        title_equiv = exercise_title_equivalent_to_stage_goal(title, stage_goal)
-        primary = (path_primary_topic or "").strip()
-        if primary and not title_equiv:
-            tech_excludes = list(path_technique_excludes or [])
-            for item in technique_sibling_excludes(primary):
-                if item not in tech_excludes:
-                    tech_excludes.append(item)
-            if not exercise_passes_technique_path_scope(
-                primary_topic=primary,
-                title=title,
-                summary=summary,
-                goal=goal_text,
-                learning_goal=stage_goal,
-                sibling_excludes=tech_excludes,
-                relaxed=True,
-            ):
-                continue
         rank_sem = float(
             hit.get("stage_rank_semantic")
             or hit.get("stage_semantic_score")
             or hit.get("semantic_score")
             or 0.0
         )
+        if not exercise_passes_stage_fit(
+            learning_goal=stage_goal,
+            title=title,
+            summary=summary,
+            goal=goal_text,
+            stage_brief=stage_brief,
+            stage_semantic_score=rank_sem,
+            anti_patterns=stage_anti_patterns,
+            path_primary_topic=path_primary_topic,
+            path_technique_excludes=path_technique_excludes,
+            relaxed=True,
+        ):
+            continue
         score = float(hit.get("score") or 0.0)
         key = (rank_sem, score)
         if key > best_key:
@@ -1427,6 +1460,7 @@ def pick_best_path_hit(
     stage_match_brief: Optional[PlanningSemanticBrief] = None,
     path_primary_topic: Optional[str] = None,
     path_technique_excludes: Optional[Sequence[str]] = None,
+    peer_learning_goals: Optional[Sequence[str]] = None,
 ) -> Optional[Dict[str, Any]]:
     """Gestufte Auswahl: strikt → relaxed → optional Notfall-Fallback."""
     if not hits:
@@ -1451,6 +1485,12 @@ def pick_best_path_hit(
             title = str(hit.get("title") or "")
             summary = str(hit.get("summary") or "")
             goal_text = str(hit.get("goal") or hit.get("exercise_goal") or "")
+            if peer_learning_goals and exercise_title_matches_peer_stage_goal(
+                title,
+                current_learning_goal=stage_goal,
+                peer_learning_goals=peer_learning_goals,
+            ):
+                continue
             sem = float(hit.get("semantic_score") or 0.0)
             stage_sem = float(
                 hit.get("stage_rank_semantic")
@@ -1506,6 +1546,8 @@ def pick_best_path_hit(
             stage_anti_patterns=stage_anti_patterns,
             path_primary_topic=path_primary_topic,
             path_technique_excludes=path_technique_excludes,
+            stage_match_brief=stage_brief,
+            peer_learning_goals=peer_learning_goals,
         )
 
     chosen = _scan(strict=False)
@@ -1546,6 +1588,7 @@ __all__ = [
     "build_stage_match_brief",
     "enrich_brief_with_path_constraints",
     "exercise_passes_stage_fit",
+    "exercise_title_matches_peer_stage_goal",
     "exercise_title_equivalent_to_stage_goal",
     "resolve_path_primary_topic",
     "resolve_path_anti_patterns",
@@ -1555,6 +1598,7 @@ __all__ = [
     "merge_stage_exclude_phrases",
     "parse_stage_goal_constraints",
     "stage_focus_phrases_from_learning_goal",
+    "stage_refinement_criteria_from_learning_goal",
     "pick_best_path_hit",
     "exercise_passes_technique_path_scope",
     "score_exercise_stage_fit",

backend/planning_path_refine_stage.py

@@ -12,7 +12,7 @@ from planning_exercise_semantics import (
     is_trainer_stage_anti_marker,
     merge_stage_exclude_phrases,
     parse_stage_goal_constraints,
-    stage_focus_phrases_from_learning_goal,
+    stage_refinement_criteria_from_learning_goal,
 )
 from planning_progression_roadmap import ProgressionRoadmapContext, StageSpecArtifact
 
@@ -118,7 +118,7 @@ def refine_stage_spec_artifact(
             anti.append(phrase)
             changes.append(f"Ausschluss aus Lernziel: {phrase[:60]}")
 
-    for phrase in stage_focus_phrases_from_learning_goal(learning_goal):
+    for phrase in stage_refinement_criteria_from_learning_goal(learning_goal):
         crit = f"Bezug zu Stufen-Lernziel: {phrase[:100]}"
         if crit not in success:
             success.append(crit)

backend/tests/test_planning_stage_anti_patterns.py

@@ -52,3 +52,70 @@ def test_stage_focus_scoring_rewards_learning_goal_tokens():
         stage_brief=brief,
     )
     assert score >= 0.25
+
+
+def test_rank_fallback_requires_relaxed_stage_fit():
+    goal_a = "Grundlegende Körperhaltung und erste Mae Geri Bewegung"
+    goal_b = "Präzise Trefferfläche und variable Distanzen"
+    hits = [
+        {
+            "id": 1,
+            "title": "Gleichgewichtstritt Mae-Geri",
+            "summary": "Balance",
+            "goal": "Mae Geri",
+            "stage_rank_semantic": 0.04,
+            "score": 0.5,
+        },
+        {
+            "id": 2,
+            "title": "Erlernen des Mae Geri aus zusammengesetzten Einzelbewegungen",
+            "summary": "Teile verbinden",
+            "goal": "Zusammensetzung",
+            "stage_rank_semantic": 0.03,
+            "score": 0.48,
+        },
+    ]
+    from planning_exercise_semantics import pick_best_path_hit
+
+    brief_a = build_stage_match_brief(learning_goal=goal_a)
+    chosen = pick_best_path_hit(
+        hits,
+        set(),
+        stage_learning_goal=goal_a,
+        roadmap_stage_match=True,
+        stage_match_brief=brief_a,
+        peer_learning_goals=[goal_b],
+    )
+    assert chosen is None
+
+
+def test_peer_stage_title_blocked_for_wrong_slot():
+    goal_a = "Grundlegende Körperhaltung und erste Mae Geri Bewegung"
+    goal_b = "Gleichgewichtstritt Mae-Geri"
+    from planning_exercise_semantics import exercise_title_matches_peer_stage_goal, pick_best_path_hit
+
+    assert exercise_title_matches_peer_stage_goal(
+        "Gleichgewichtstritt Mae-Geri",
+        current_learning_goal=goal_a,
+        peer_learning_goals=[goal_b],
+    )
+    hits = [
+        {
+            "id": 10,
+            "title": "Gleichgewichtstritt Mae-Geri",
+            "summary": "Balance auf einem Bein",
+            "goal": "Mae Geri aus Gleichgewicht",
+            "stage_rank_semantic": 0.35,
+            "score": 0.6,
+        }
+    ]
+    brief_a = build_stage_match_brief(learning_goal=goal_a)
+    chosen = pick_best_path_hit(
+        hits,
+        set(),
+        stage_learning_goal=goal_a,
+        roadmap_stage_match=True,
+        stage_match_brief=brief_a,
+        peer_learning_goals=[goal_b],
+    )
+    assert chosen is None

backend/version.py

@@ -1,6 +1,6 @@
 # Shinkan Jinkendo Version Information
 
-APP_VERSION = "0.8.228"
+APP_VERSION = "0.8.229"
 BUILD_DATE = "2026-05-22"
 DB_SCHEMA_VERSION = "20260607090"
 
@@ -38,7 +38,7 @@ MODULE_VERSIONS = {
     "skill_profiles": "1.0.0",  # Phase 3: gewichtetes Fähigkeiten-Profil + skill-discovery/suggestions
     "methods": "0.1.0",
     "exercises": "2.37.1",  # KI-Endpoints: feature_usage nach ai_calls consume
-    "planning_exercise_suggest": "0.23.3",  # Stufen-Match: saubere Anti-Patterns, Fit-Scoring, Rematch-Akkumulation
+    "planning_exercise_suggest": "0.23.4",  # Stufen-Match: Fallback mit Gate, Peer-Slot-Schutz, LG-Kandidaten-Filter
     "training_units": "0.4.0",  # POST .../publish-to-framework: Ablauf aus geplanter Einheit → Rahmen-Slot-Blueprint
     "training_programs": "0.1.0",
     "planning": "0.15.0",  # Vorlagen: Strukturvorschau, Bearbeiten inkl. Split-Sessions + Beschreibung