From 1d94c2ebf136859d1c6e2ce723b02f32b2ef104a Mon Sep 17 00:00:00 2001 From: Lars Date: Thu, 11 Jun 2026 10:30:48 +0200 Subject: [PATCH] Enhance Roadmap Slot Matching and Off-Topic Detection - Introduced `auto_rematch_after_qa` parameter in `ProgressionPathSuggestRequest` to enable automatic rematching after quality assurance checks. - Refactored roadmap slot matching logic to improve clarity and functionality, renaming `_build_steps_roadmap_first` to `_match_roadmap_slot`. - Added `_with_roadmap_major_index` utility to streamline off-topic step detection by incorporating roadmap major step indices. - Enhanced off-topic detection logic to utilize the new utility for improved clarity in identifying mismatches and exclusions. - Incremented application version to reflect these updates. --- backend/planning_exercise_path_builder.py | 377 +++++++++++++------- backend/planning_exercise_path_qa.py | 96 +++-- backend/planning_path_rematch.py | 202 +++++++++++ backend/tests/test_planning_path_rematch.py | 133 +++++++ 4 files changed, 635 insertions(+), 173 deletions(-) create mode 100644 backend/planning_path_rematch.py create mode 100644 backend/tests/test_planning_path_rematch.py diff --git a/backend/planning_exercise_path_builder.py b/backend/planning_exercise_path_builder.py index 9a56cf5..7a9b4c7 100644 --- a/backend/planning_exercise_path_builder.py +++ b/backend/planning_exercise_path_builder.py @@ -14,6 +14,7 @@ from pydantic import BaseModel, Field from tenant_context import TenantContext, library_content_visibility_sql from planning_exercise_profiles import PlanningTargetProfile from planning_path_qa_pipeline import run_multistage_path_qa +from planning_path_rematch import collect_rematch_slot_indices, rematch_roadmap_slots from planning_stage_context import build_contextualized_stage_goal, resolve_path_start_target from planning_exercise_path_qa import ( apply_llm_path_reorder, @@ -97,6 +98,7 @@ class ProgressionPathSuggestRequest(BaseModel): max_steps: int = Field(default=5, ge=2, le=10) include_llm_intent: bool = True include_path_qa: bool = True + auto_rematch_after_qa: bool = True include_llm_path_qa: bool = True include_path_reorder: bool = True include_ai_gap_fill: bool = True @@ -534,6 +536,174 @@ def _annotate_roadmap_step( return step +def _match_roadmap_slot( + cur, + *, + tenant: TenantContext, + body: ProgressionPathSuggestRequest, + goal_query: str, + max_steps: int, + semantic_brief: PlanningSemanticBrief, + path_target_profile: PlanningTargetProfile, + path_intent: str, + roadmap_ctx: ProgressionRoadmapContext, + stage_spec: StageSpecArtifact, + step_index: int, + stage_count: int, + planned_ids: List[int], + anchor_id: Optional[int], + anchor_variant_id: Optional[int], + used: Set[int], +) -> Tuple[Optional[Dict[str, Any]], Optional[StageSpecArtifact]]: + """Einzelnen Roadmap-Slot matchen (Initial-Build und Auto-Rematch).""" + major_by_index: Dict[int, MajorStep] = {} + if roadmap_ctx.roadmap: + major_by_index = {m.index: m for m in roadmap_ctx.roadmap.major_steps} + major = major_by_index.get(stage_spec.major_step_index) + + ga_dump = ( + roadmap_ctx.goal_analysis.model_dump() if roadmap_ctx.goal_analysis else None + ) + rs_dump = ( + roadmap_ctx.resolved_structured.model_dump() + if roadmap_ctx.resolved_structured + else None + ) + path_start, path_target = resolve_path_start_target( + structured=roadmap_ctx.resolved_structured, + goal_analysis=roadmap_ctx.goal_analysis, + ) + brief_summary = ( + roadmap_ctx.semantic_brief + if roadmap_ctx.semantic_brief + else brief_to_summary_dict(semantic_brief) + ) + + stage_spec_dict = stage_spec.model_dump() + if major: + stage_spec_dict["phase"] = major.phase + stage_inp = expectation_input_from_progression_stage( + goal_query=goal_query, + goal_analysis=ga_dump, + resolved_structured=rs_dump, + stage_spec=stage_spec_dict, + semantic_brief_summary=brief_summary, + major_step=major.model_dump() if major else None, + ) + stage_exp = build_planning_skill_expectations(cur, stage_inp, semantic_brief=semantic_brief) + step_target = apply_expectations_to_target(path_target_profile, stage_exp) + skill_exp_api = stage_exp.to_api_dict() if stage_exp.items else None + + step_query = stage_spec_retrieval_query( + semantic_brief=semantic_brief, + goal_query=goal_query, + stage_spec=stage_spec, + major_step=major, + ) + step_kind = resolve_step_exercise_kind_filter(stage_spec, body.exercise_kind_any) + stage_goal = (stage_spec.learning_goal or "").strip() + stage_start = (stage_spec.start_state or "").strip() + stage_target = (stage_spec.target_state or "").strip() + contextual_goal = build_contextualized_stage_goal( + learning_goal=stage_goal, + start_state=stage_start, + target_state=stage_target, + path_target_state=path_target, + path_start_state=path_start, + stage_index=step_index, + stage_count=stage_count, + ) + path_context_note = None + if rs_dump: + ctx_parts = [ + str(rs_dump.get("start_situation") or "").strip()[:120], + str(rs_dump.get("target_state") or "").strip()[:120], + str(rs_dump.get("roadmap_notes") or "").strip()[:120], + ] + path_context_note = " ".join(p for p in ctx_parts if p)[:240] or None + path_anti = resolve_path_anti_patterns( + goal_query, + semantic_brief=semantic_brief, + extra_context=path_context_note, + ) + stage_anti = list(dict.fromkeys([*(stage_spec.anti_patterns or []), *path_anti])) + path_primary = (semantic_brief.primary_topic or "").strip() + path_tech_excludes = list(semantic_brief.exclude_phrases or []) + if semantic_brief.topic_type == "technique" and path_primary: + from planning_exercise_semantics import technique_sibling_excludes + + for item in technique_sibling_excludes(path_primary): + if item not in path_tech_excludes: + path_tech_excludes.append(item) + stage_match_brief = build_stage_match_brief( + learning_goal=stage_goal, + anti_patterns=stage_anti, + success_criteria=list(stage_spec.success_criteria or []), + load_profile=list(stage_spec.load_profile or []), + phase=major.phase if major else None, + path_context_note=path_context_note, + path_anti_patterns=path_anti, + path_primary_topic=path_primary or None, + path_technique_excludes=path_tech_excludes or None, + stage_start_state=stage_start or None, + stage_target_state=stage_target or None, + path_target_state=path_target or None, + contextualized_learning_goal=contextual_goal or None, + ) + + hits, _, _, _ = _run_path_step_retrieval( + cur, + tenant=tenant, + goal_query=goal_query, + step_index=step_index, + max_steps=max_steps, + planned_ids=planned_ids, + anchor_id=anchor_id, + anchor_variant_id=anchor_variant_id, + progression_graph_id=body.progression_graph_id, + include_llm_intent=body.include_llm_intent and step_index == 0, + exercise_kind_any=step_kind, + semantic_brief=stage_match_brief, + path_target_profile=path_target_profile, + path_intent=path_intent, + step_query_override=step_query, + step_phase_override=major.phase if major else None, + step_target_profile_override=step_target, + stage_learning_goal=stage_goal or None, + stage_anti_patterns=stage_anti or None, + stage_match_brief=stage_match_brief, + stage_success_criteria=list(stage_spec.success_criteria or []), + stage_load_profile=list(stage_spec.load_profile or []), + path_context_note=path_context_note, + path_primary_topic=path_primary or None, + path_technique_excludes=path_tech_excludes or None, + ) + + hit = _pick_best_path_hit( + hits, + used, + semantic_brief=stage_match_brief, + stage_learning_goal=stage_goal or None, + stage_anti_patterns=stage_anti or None, + roadmap_stage_match=True, + stage_match_brief=stage_match_brief, + path_primary_topic=path_primary or None, + path_technique_excludes=path_tech_excludes or None, + ) + + if not hit: + return None, stage_spec + + step = _annotate_roadmap_step( + _hit_to_path_step(hit), + stage_spec=stage_spec, + major_step=major, + skill_expectations=skill_exp_api, + anti_patterns_override=stage_anti, + ) + return step, None + + def _build_steps_roadmap_first( cur, *, @@ -557,161 +727,37 @@ def _build_steps_roadmap_first( for m in roadmap_ctx.roadmap.major_steps[:max_steps] ] - major_by_index: Dict[int, MajorStep] = {} - if roadmap_ctx.roadmap: - major_by_index = {m.index: m for m in roadmap_ctx.roadmap.major_steps} - used: Set[int] = set() steps: List[Dict[str, Any]] = [] planned_ids: List[int] = [] anchor_id: Optional[int] = None anchor_variant_id: Optional[int] = None unfilled: List[Tuple[int, StageSpecArtifact]] = [] - - ga_dump = ( - roadmap_ctx.goal_analysis.model_dump() if roadmap_ctx.goal_analysis else None - ) - rs_dump = ( - roadmap_ctx.resolved_structured.model_dump() - if roadmap_ctx.resolved_structured - else None - ) - path_start, path_target = resolve_path_start_target( - structured=roadmap_ctx.resolved_structured, - goal_analysis=roadmap_ctx.goal_analysis, - ) stage_count = len(stage_specs) - brief_summary = ( - roadmap_ctx.semantic_brief - if roadmap_ctx.semantic_brief - else brief_to_summary_dict(semantic_brief) - ) for step_index, stage_spec in enumerate(stage_specs): - major = major_by_index.get(stage_spec.major_step_index) - stage_spec_dict = stage_spec.model_dump() - if major: - stage_spec_dict["phase"] = major.phase - stage_inp = expectation_input_from_progression_stage( - goal_query=goal_query, - goal_analysis=ga_dump, - resolved_structured=rs_dump, - stage_spec=stage_spec_dict, - semantic_brief_summary=brief_summary, - major_step=major.model_dump() if major else None, - ) - stage_exp = build_planning_skill_expectations(cur, stage_inp, semantic_brief=semantic_brief) - step_target = apply_expectations_to_target(path_target_profile, stage_exp) - skill_exp_api = stage_exp.to_api_dict() if stage_exp.items else None - - step_query = stage_spec_retrieval_query( - semantic_brief=semantic_brief, - goal_query=goal_query, - stage_spec=stage_spec, - major_step=major, - ) - step_kind = resolve_step_exercise_kind_filter(stage_spec, body.exercise_kind_any) - stage_goal = (stage_spec.learning_goal or "").strip() - stage_start = (stage_spec.start_state or "").strip() - stage_target = (stage_spec.target_state or "").strip() - contextual_goal = build_contextualized_stage_goal( - learning_goal=stage_goal, - start_state=stage_start, - target_state=stage_target, - path_target_state=path_target, - path_start_state=path_start, - stage_index=step_index, - stage_count=stage_count, - ) - path_context_note = None - if rs_dump: - ctx_parts = [ - str(rs_dump.get("start_situation") or "").strip()[:120], - str(rs_dump.get("target_state") or "").strip()[:120], - str(rs_dump.get("roadmap_notes") or "").strip()[:120], - ] - path_context_note = " ".join(p for p in ctx_parts if p)[:240] or None - path_anti = resolve_path_anti_patterns( - goal_query, - semantic_brief=semantic_brief, - extra_context=path_context_note, - ) - stage_anti = list(dict.fromkeys([*(stage_spec.anti_patterns or []), *path_anti])) - path_primary = (semantic_brief.primary_topic or "").strip() - path_tech_excludes = list(semantic_brief.exclude_phrases or []) - if semantic_brief.topic_type == "technique" and path_primary: - from planning_exercise_semantics import technique_sibling_excludes - - for item in technique_sibling_excludes(path_primary): - if item not in path_tech_excludes: - path_tech_excludes.append(item) - stage_match_brief = build_stage_match_brief( - learning_goal=stage_goal, - anti_patterns=stage_anti, - success_criteria=list(stage_spec.success_criteria or []), - load_profile=list(stage_spec.load_profile or []), - phase=major.phase if major else None, - path_context_note=path_context_note, - path_anti_patterns=path_anti, - path_primary_topic=path_primary or None, - path_technique_excludes=path_tech_excludes or None, - stage_start_state=stage_start or None, - stage_target_state=stage_target or None, - path_target_state=path_target or None, - contextualized_learning_goal=contextual_goal or None, - ) - - hits, _, _, _ = _run_path_step_retrieval( + step, unfilled_spec = _match_roadmap_slot( cur, tenant=tenant, + body=body, goal_query=goal_query, - step_index=step_index, max_steps=max_steps, + semantic_brief=semantic_brief, + path_target_profile=path_target_profile, + path_intent=path_intent, + roadmap_ctx=roadmap_ctx, + stage_spec=stage_spec, + step_index=step_index, + stage_count=stage_count, planned_ids=planned_ids, anchor_id=anchor_id, anchor_variant_id=anchor_variant_id, - progression_graph_id=body.progression_graph_id, - include_llm_intent=body.include_llm_intent and step_index == 0, - exercise_kind_any=step_kind, - semantic_brief=stage_match_brief, - path_target_profile=path_target_profile, - path_intent=path_intent, - step_query_override=step_query, - step_phase_override=major.phase if major else None, - step_target_profile_override=step_target, - stage_learning_goal=stage_goal or None, - stage_anti_patterns=stage_anti or None, - stage_match_brief=stage_match_brief, - stage_success_criteria=list(stage_spec.success_criteria or []), - stage_load_profile=list(stage_spec.load_profile or []), - path_context_note=path_context_note, - path_primary_topic=path_primary or None, - path_technique_excludes=path_tech_excludes or None, + used=used, ) - - hit = _pick_best_path_hit( - hits, - used, - semantic_brief=stage_match_brief, - stage_learning_goal=stage_goal or None, - stage_anti_patterns=stage_anti or None, - roadmap_stage_match=True, - stage_match_brief=stage_match_brief, - path_primary_topic=path_primary or None, - path_technique_excludes=path_tech_excludes or None, - ) - - if not hit: - unfilled.append((step_index, stage_spec)) + if not step: + unfilled.append((step_index, unfilled_spec or stage_spec)) continue - step = _annotate_roadmap_step( - _hit_to_path_step(hit), - stage_spec=stage_spec, - major_step=major, - skill_expectations=skill_exp_api, - anti_patterns_override=stage_anti, - ) steps.append(step) eid = int(step["exercise_id"]) used.add(eid) @@ -1245,6 +1291,8 @@ def suggest_progression_path( gap_fill_offers: List[Dict[str, Any]] = [] off_topic_steps: List[Dict[str, Any]] = [] stripped_off_topic: List[Dict[str, Any]] = [] + rematch_log: List[Dict[str, Any]] = [] + rematch_rounds = 0 llm_qa: Optional[Dict[str, Any]] = None llm_qa_applied = False reorder_applied = False @@ -1315,6 +1363,7 @@ def suggest_progression_path( brief=semantic_brief, goal_query=goal_query, ) + off_topic_before_strip = list(off_topic_steps) steps, stripped_off_topic = strip_off_topic_steps_from_path(steps, off_topic_steps) if stripped_off_topic: off_topic_steps = [] @@ -1325,6 +1374,56 @@ def suggest_progression_path( roadmap_first=roadmap_first, ) + if ( + roadmap_first + and body.auto_rematch_after_qa + and roadmap_ctx is not None + and roadmap_ctx.stage_specs + ): + slot_indices, rematch_reasons = collect_rematch_slot_indices( + stripped_off_topic=stripped_off_topic, + off_topic_steps=off_topic_before_strip if not stripped_off_topic else [], + optimization_hints=[], + stage_specs=roadmap_ctx.stage_specs, + ) + if slot_indices: + steps, rematch_log, rematch_new_unfilled = rematch_roadmap_slots( + cur, + tenant=tenant, + body=body, + goal_query=goal_query, + max_steps=max_steps, + semantic_brief=semantic_brief, + path_target_profile=path_target_profile, + path_intent=path_intent, + roadmap_ctx=roadmap_ctx, + steps=steps, + slot_indices=slot_indices, + rematch_reasons=rematch_reasons, + match_slot_fn=_match_roadmap_slot, + ) + rematch_rounds = 1 + if rematch_new_unfilled: + remapped = {sp.major_step_index for _, sp in rematch_new_unfilled} + roadmap_unfilled = [ + item + for item in roadmap_unfilled + if item[1].major_step_index not in remapped + ] + roadmap_unfilled.extend(rematch_new_unfilled) + off_topic_steps = detect_off_topic_steps( + cur, + steps, + brief=semantic_brief, + goal_query=goal_query, + ) + gaps = detect_path_gaps( + cur, + steps, + brief=semantic_brief, + roadmap_first=roadmap_first, + ) + llm_gap_specs = parse_llm_suggested_new_exercises( llm_qa, brief=semantic_brief, @@ -1396,6 +1495,10 @@ def suggest_progression_path( roadmap_qa_mode=roadmap_qa_mode, multistage_qa=multistage_qa, ) + if rematch_log: + path_qa["rematch_applied"] = True + path_qa["rematch_log"] = rematch_log + path_qa["rematch_rounds"] = rematch_rounds target_profile_summary = path_target_profile.to_summary_dict(cur) retrieval_parts = ["profile_v1", "full_library", "path_builder", "semantics"] @@ -1419,6 +1522,8 @@ def suggest_progression_path( retrieval_parts.append("roadmap_edited") if roadmap_unfilled: retrieval_parts.append("roadmap_unfilled") + if rematch_log: + retrieval_parts.append("path_rematch") return { "goal_query": goal_query, diff --git a/backend/planning_exercise_path_qa.py b/backend/planning_exercise_path_qa.py index 4ce77ba..8b833a4 100644 --- a/backend/planning_exercise_path_qa.py +++ b/backend/planning_exercise_path_qa.py @@ -398,6 +398,16 @@ def apply_llm_path_reorder( _OFF_TOPIC_SEMANTIC_MAX = 0.10 +def _with_roadmap_major_index( + step: Mapping[str, Any], + entry: Dict[str, Any], +) -> Dict[str, Any]: + midx = step.get("roadmap_major_step_index") + if midx is not None: + entry["roadmap_major_step_index"] = int(midx) + return entry + + def detect_off_topic_steps( cur, steps: Sequence[Mapping[str, Any]], @@ -425,15 +435,18 @@ def detect_off_topic_steps( step_anti = list(step.get("roadmap_anti_patterns") or []) + path_anti if step_anti and _blob_matches_stage_excludes(blob, step_anti): off_topic.append( - { - "step_index": idx, - "exercise_id": int(step["exercise_id"]), - "title": step.get("title") or bundle["title"], - "semantic_score": 0.0, - "expected_phase": (step.get("roadmap_phase") or "").strip().lower() or None, - "issue": "path_exclude", - "reasons": ["Widerspricht Pfad-Ausschlüssen (z. B. Kumite)"], - } + _with_roadmap_major_index( + step, + { + "step_index": idx, + "exercise_id": int(step["exercise_id"]), + "title": step.get("title") or bundle["title"], + "semantic_score": 0.0, + "expected_phase": (step.get("roadmap_phase") or "").strip().lower() or None, + "issue": "path_exclude", + "reasons": ["Widerspricht Pfad-Ausschlüssen (z. B. Kumite)"], + }, + ) ) continue primary = (brief.primary_topic or "").strip() @@ -450,15 +463,18 @@ def detect_off_topic_steps( relaxed=False, ): off_topic.append( - { - "step_index": idx, - "exercise_id": int(step["exercise_id"]), - "title": step.get("title") or bundle["title"], - "semantic_score": 0.0, - "expected_phase": (step.get("roadmap_phase") or "").strip().lower() or None, - "issue": "technique_scope", - "reasons": [f"Passt nicht zur Haupttechnik „{primary}“"], - } + _with_roadmap_major_index( + step, + { + "step_index": idx, + "exercise_id": int(step["exercise_id"]), + "title": step.get("title") or bundle["title"], + "semantic_score": 0.0, + "expected_phase": (step.get("roadmap_phase") or "").strip().lower() or None, + "issue": "technique_scope", + "reasons": [f"Passt nicht zur Haupttechnik „{primary}“"], + }, + ) ) continue stage_goal = (step.get("roadmap_learning_goal") or "").strip() @@ -488,16 +504,19 @@ def detect_off_topic_steps( anti_patterns=stage_anti or None, ): off_topic.append( - { - "step_index": idx, - "exercise_id": int(step["exercise_id"]), - "title": step.get("title") or bundle["title"], - "semantic_score": round(sem, 4), - "expected_phase": phase, - "issue": "stage_mismatch", - "roadmap_learning_goal": stage_goal, - "reasons": sem_reasons[:3], - } + _with_roadmap_major_index( + step, + { + "step_index": idx, + "exercise_id": int(step["exercise_id"]), + "title": step.get("title") or bundle["title"], + "semantic_score": round(sem, 4), + "expected_phase": phase, + "issue": "stage_mismatch", + "roadmap_learning_goal": stage_goal, + "reasons": sem_reasons[:3], + }, + ) ) continue if exercise_passes_path_semantic_gate( @@ -512,15 +531,18 @@ def detect_off_topic_steps( if sem > _OFF_TOPIC_SEMANTIC_MAX: continue off_topic.append( - { - "step_index": idx, - "exercise_id": int(step["exercise_id"]), - "title": step.get("title") or bundle["title"], - "semantic_score": round(sem, 4), - "expected_phase": phase, - "issue": "off_topic", - "reasons": sem_reasons[:3], - } + _with_roadmap_major_index( + step, + { + "step_index": idx, + "exercise_id": int(step["exercise_id"]), + "title": step.get("title") or bundle["title"], + "semantic_score": round(sem, 4), + "expected_phase": phase, + "issue": "off_topic", + "reasons": sem_reasons[:3], + }, + ) ) return off_topic diff --git a/backend/planning_path_rematch.py b/backend/planning_path_rematch.py new file mode 100644 index 0000000..1faeba8 --- /dev/null +++ b/backend/planning_path_rematch.py @@ -0,0 +1,202 @@ +""" +Auto-Rematch nach Pfad-QS — betroffene Roadmap-Slots erneut matchen (Phase A). +""" +from __future__ import annotations + +from typing import Any, Dict, List, Mapping, Optional, Sequence, Set, Tuple + +from planning_progression_roadmap import ProgressionRoadmapContext, StageSpecArtifact + + +def collect_rematch_slot_indices( + *, + stripped_off_topic: Sequence[Mapping[str, Any]], + off_topic_steps: Sequence[Mapping[str, Any]], + optimization_hints: Sequence[Mapping[str, Any]], + stage_specs: Sequence[StageSpecArtifact], +) -> Tuple[Set[int], Dict[int, str]]: + """Major-Step-Indizes für rematch_slot + Begründung pro Slot.""" + spec_by_pos = list(stage_specs) + indices: Set[int] = set() + reasons: Dict[int, str] = {} + + def _register(midx: int, reason: str) -> None: + indices.add(int(midx)) + if midx not in reasons and reason: + reasons[int(midx)] = reason[:400] + + def _resolve_major(item: Mapping[str, Any]) -> Optional[int]: + raw = item.get("roadmap_major_step_index") + if raw is not None: + return int(raw) + si = item.get("step_index") + if si is not None: + pos = int(si) + if 0 <= pos < len(spec_by_pos): + return int(spec_by_pos[pos].major_step_index) + return None + + for item in stripped_off_topic or []: + if not isinstance(item, dict): + continue + midx = _resolve_major(item) + if midx is not None: + issue = str(item.get("issue") or "stripped_off_topic") + r = (item.get("reasons") or [issue])[0] if item.get("reasons") else issue + _register(midx, str(r)) + + for item in off_topic_steps or []: + if not isinstance(item, dict): + continue + midx = _resolve_major(item) + if midx is None: + continue + issue = str(item.get("issue") or "off_topic") + r = (item.get("reasons") or [issue])[0] if item.get("reasons") else issue + _register(midx, str(r)) + + for hint in optimization_hints or []: + if not isinstance(hint, dict): + continue + if str(hint.get("action") or "") != "rematch_slot": + continue + midx = _resolve_major(hint) + if midx is not None: + _register(midx, str(hint.get("reason") or hint.get("issue") or "rematch_slot")) + + return indices, reasons + + +def _context_before_major( + steps_by_major: Mapping[int, Mapping[str, Any]], + target_major: int, +) -> Tuple[List[int], Optional[int], Optional[int]]: + planned: List[int] = [] + anchor: Optional[int] = None + anchor_vid: Optional[int] = None + for midx in sorted(steps_by_major): + if midx >= target_major: + break + step = steps_by_major[midx] + eid = step.get("exercise_id") + if eid is not None: + planned.append(int(eid)) + anchor = int(eid) + vid = step.get("variant_id") + anchor_vid = int(vid) if vid is not None else None + return planned, anchor, anchor_vid + + +def rematch_roadmap_slots( + cur, + *, + tenant, + body, + goal_query: str, + max_steps: int, + semantic_brief, + path_target_profile, + path_intent: str, + roadmap_ctx: ProgressionRoadmapContext, + steps: Sequence[Mapping[str, Any]], + slot_indices: Set[int], + rematch_reasons: Mapping[int, str], + match_slot_fn, +) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], List[Tuple[int, StageSpecArtifact]]]: + """ + Ersetzt nur betroffene Slots; andere Schritte und used-Set bleiben konsistent. + + match_slot_fn: _match_roadmap_slot aus path_builder (Injection gegen Zirkularität). + """ + stage_specs = list(roadmap_ctx.stage_specs or [])[:max_steps] + if not stage_specs or not slot_indices: + return list(steps), [], [] + + spec_by_major = {int(s.major_step_index): s for s in stage_specs} + steps_by_major: Dict[int, Dict[str, Any]] = {} + for raw in steps: + step = dict(raw) + midx = step.get("roadmap_major_step_index") + if midx is not None: + steps_by_major[int(midx)] = step + + rematch_log: List[Dict[str, Any]] = [] + new_unfilled: List[Tuple[int, StageSpecArtifact]] = [] + + for major_idx in sorted(slot_indices): + stage_spec = spec_by_major.get(int(major_idx)) + if stage_spec is None: + continue + step_index = next( + (i for i, sp in enumerate(stage_specs) if int(sp.major_step_index) == int(major_idx)), + major_idx, + ) + old = steps_by_major.pop(int(major_idx), None) + used = { + int(s["exercise_id"]) + for m, s in steps_by_major.items() + if s.get("exercise_id") is not None + } + planned_ids, anchor_id, anchor_variant_id = _context_before_major( + steps_by_major, int(major_idx) + ) + + new_step, unfilled_spec = match_slot_fn( + cur, + tenant=tenant, + body=body, + goal_query=goal_query, + max_steps=max_steps, + semantic_brief=semantic_brief, + path_target_profile=path_target_profile, + path_intent=path_intent, + roadmap_ctx=roadmap_ctx, + stage_spec=stage_spec, + step_index=step_index, + stage_count=len(stage_specs), + planned_ids=planned_ids, + anchor_id=anchor_id, + anchor_variant_id=anchor_variant_id, + used=used, + ) + + reason = str(rematch_reasons.get(int(major_idx)) or "rematch_slot") + if new_step: + steps_by_major[int(major_idx)] = new_step + rematch_log.append( + { + "roadmap_major_step_index": int(major_idx), + "action": "replaced", + "reason": reason, + "replaced_exercise_id": old.get("exercise_id") if old else None, + "replaced_title": old.get("title") if old else None, + "new_exercise_id": new_step.get("exercise_id"), + "new_title": new_step.get("title"), + } + ) + else: + if unfilled_spec is not None: + new_unfilled.append((step_index, unfilled_spec)) + rematch_log.append( + { + "roadmap_major_step_index": int(major_idx), + "action": "rematch_unfilled", + "reason": reason, + "replaced_exercise_id": old.get("exercise_id") if old else None, + "replaced_title": old.get("title") if old else None, + } + ) + + ordered: List[Dict[str, Any]] = [] + for spec in sorted(stage_specs, key=lambda s: s.major_step_index): + midx = int(spec.major_step_index) + if midx in steps_by_major: + ordered.append(steps_by_major[midx]) + + return ordered, rematch_log, new_unfilled + + +__all__ = [ + "collect_rematch_slot_indices", + "rematch_roadmap_slots", +] diff --git a/backend/tests/test_planning_path_rematch.py b/backend/tests/test_planning_path_rematch.py new file mode 100644 index 0000000..fbacba9 --- /dev/null +++ b/backend/tests/test_planning_path_rematch.py @@ -0,0 +1,133 @@ +"""Tests Auto-Rematch nach Pfad-QS (Phase A).""" +from planning_path_rematch import collect_rematch_slot_indices, rematch_roadmap_slots +from planning_progression_roadmap import ProgressionRoadmapContext, StageSpecArtifact + + +def _stage_specs(): + return [ + StageSpecArtifact(major_step_index=0, learning_goal="Grundlage"), + StageSpecArtifact(major_step_index=1, learning_goal="Vertiefung"), + StageSpecArtifact(major_step_index=2, learning_goal="Anwendung"), + ] + + +def test_collect_rematch_slot_indices_from_stripped_with_major_index(): + specs = _stage_specs() + stripped = [ + { + "step_index": 1, + "roadmap_major_step_index": 1, + "issue": "technique_scope", + "reasons": ["Passt nicht zur Haupttechnik"], + } + ] + indices, reasons = collect_rematch_slot_indices( + stripped_off_topic=stripped, + off_topic_steps=[], + optimization_hints=[], + stage_specs=specs, + ) + assert indices == {1} + assert "Haupttechnik" in reasons[1] + + +def test_collect_rematch_slot_indices_resolves_step_index_to_major(): + specs = _stage_specs() + off_topic = [ + { + "step_index": 2, + "issue": "stage_mismatch", + "reasons": ["Ziel passt nicht"], + } + ] + indices, reasons = collect_rematch_slot_indices( + stripped_off_topic=[], + off_topic_steps=off_topic, + optimization_hints=[], + stage_specs=specs, + ) + assert indices == {2} + assert reasons[2] == "Ziel passt nicht" + + +def test_collect_rematch_slot_indices_from_optimization_hints(): + specs = _stage_specs() + hints = [ + { + "action": "rematch_slot", + "roadmap_major_step_index": 0, + "reason": "QS-Tier-1", + } + ] + indices, _ = collect_rematch_slot_indices( + stripped_off_topic=[], + off_topic_steps=[], + optimization_hints=hints, + stage_specs=specs, + ) + assert indices == {0} + + +def test_rematch_roadmap_slots_replaces_only_target_slot(): + specs = _stage_specs() + ctx = ProgressionRoadmapContext( + goal_query="Mawashi Geri", + max_steps=3, + stage_specs=specs, + ) + steps = [ + { + "exercise_id": 10, + "title": "Slot 0 OK", + "roadmap_major_step_index": 0, + }, + { + "exercise_id": 20, + "title": "Mae Geri falsch", + "roadmap_major_step_index": 1, + }, + { + "exercise_id": 30, + "title": "Slot 2 OK", + "roadmap_major_step_index": 2, + }, + ] + + def _fake_match(cur, *, stage_spec, used, **kwargs): + assert stage_spec.major_step_index == 1 + assert 20 not in used + assert 10 in used + return ( + { + "exercise_id": 21, + "title": "Sprungkraft Mawashi", + "roadmap_major_step_index": 1, + }, + None, + ) + + ordered, log, unfilled = rematch_roadmap_slots( + None, + tenant=None, + body=None, + goal_query="Mawashi Geri", + max_steps=3, + semantic_brief=None, + path_target_profile=None, + path_intent="", + roadmap_ctx=ctx, + steps=steps, + slot_indices={1}, + rematch_reasons={1: "technique_scope"}, + match_slot_fn=_fake_match, + ) + + assert len(ordered) == 3 + assert ordered[0]["exercise_id"] == 10 + assert ordered[1]["exercise_id"] == 21 + assert ordered[2]["exercise_id"] == 30 + assert len(log) == 1 + assert log[0]["action"] == "replaced" + assert log[0]["replaced_exercise_id"] == 20 + assert log[0]["new_exercise_id"] == 21 + assert not unfilled