Lars/shinkan-jinkendo
1
0
Fork 0
Code Issues 7 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Enhance Roadmap Slot Matching and Off-Topic Detection
All checks were successful
Deploy Development / deploy (push) Successful in 44s
Details
Test Suite / pytest-backend (push) Successful in 45s
Details
Test Suite / lint-backend (push) Successful in 0s
Details
Test Suite / build-frontend (push) Successful in 13s
Details
Test Suite / k6 /health Baseline (push) Successful in 36s
Details
Test Suite / playwright-tests (push) Successful in 1m23s
Details

Browse Source 
- 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.
This commit is contained in:
Lars 2026-06-11 10:30:48 +02:00
parent a152218c45
commit 1d94c2ebf1
4 changed files with 635 additions and 173 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

377
backend/planning_exercise_path_builder.py
View File

@ -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,

96
backend/planning_exercise_path_qa.py
View File

@ -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

202
backend/planning_path_rematch.py Normal file
View File

@ -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",
]

133
backend/tests/test_planning_path_rematch.py Normal file
View File

@ -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