Lars
63c99b0ec5
All checks were successful
Deploy Development / deploy (push) Successful in 45s
Test Suite / pytest-backend (push) Successful in 45s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 13s
Test Suite / k6 /health Baseline (push) Successful in 33s
Test Suite / playwright-tests (push) Successful in 1m22s
- Introduced `_roadmap_step_passes_post_match_gate` to validate steps after matching, ensuring only relevant steps proceed. - Enhanced `_enrich_roadmap_unfilled_gap_offers` to generate AI gap offers for unfilled roadmap slots, improving exercise suggestions. - Updated `suggest_progression_path` to incorporate new gap offer logic and streamline the handling of roadmap steps. - Refined frontend logic in `applyMatchStepsToSlots` to better manage step assignments and improve clarity in slot handling. - Bumped version to 0.8.231 to reflect the new features and improvements.
284 lines
10 KiB
Python
284 lines
10 KiB
Python
"""
|
|
Auto-Rematch nach Pfad-QS — betroffene Roadmap-Slots erneut matchen (Phase A/B).
|
|
"""
|
|
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],
|
|
roadmap_unfilled: Optional[Sequence[Any]] = None,
|
|
) -> 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"))
|
|
|
|
for item in roadmap_unfilled or []:
|
|
if isinstance(item, (list, tuple)) and len(item) >= 2:
|
|
idx, spec = item[0], item[1]
|
|
midx = getattr(spec, "major_step_index", idx)
|
|
_register(int(midx), "Keine passende Übung für Roadmap-Stufe")
|
|
elif isinstance(item, dict):
|
|
midx = _resolve_major(item)
|
|
if midx is not None:
|
|
issue = str(item.get("issue") or "roadmap_unfilled")
|
|
r = (item.get("reasons") or [issue])[0] if item.get("reasons") else issue
|
|
_register(midx, str(r))
|
|
|
|
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,
|
|
rejected_by_major: Optional[Mapping[int, Set[int]]] = None,
|
|
slot_assignment_history: Optional[Mapping[int, Set[int]]] = None,
|
|
) -> 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
|
|
}
|
|
if old and old.get("exercise_id") is not None:
|
|
used.add(int(old["exercise_id"]))
|
|
for rejected_id in rejected_by_major.get(int(major_idx), set()) if rejected_by_major else set():
|
|
if rejected_id > 0:
|
|
used.add(int(rejected_id))
|
|
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:
|
|
try:
|
|
new_eid = int(new_step.get("exercise_id") or 0)
|
|
except (TypeError, ValueError):
|
|
new_eid = 0
|
|
hist = (
|
|
slot_assignment_history.get(int(major_idx), set())
|
|
if slot_assignment_history
|
|
else set()
|
|
)
|
|
if new_eid > 0 and new_eid in hist:
|
|
new_step = None
|
|
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:
|
|
goal = (stage_spec.learning_goal or "").strip()
|
|
major = None
|
|
if roadmap_ctx.roadmap:
|
|
major = next(
|
|
(m for m in roadmap_ctx.roadmap.major_steps if int(m.index) == int(major_idx)),
|
|
None,
|
|
)
|
|
steps_by_major[int(major_idx)] = {
|
|
"exercise_id": None,
|
|
"variant_id": None,
|
|
"title": goal or f"Slot {major_idx + 1}",
|
|
"is_ai_proposal": False,
|
|
"roadmap_major_step_index": int(major_idx),
|
|
"roadmap_phase": major.phase if major else None,
|
|
"roadmap_learning_goal": goal or None,
|
|
"roadmap_match_source": "unfilled",
|
|
"slot_status": "unfilled",
|
|
"reasons": ["Keine passende Übung für Roadmap-Stufe"],
|
|
}
|
|
if unfilled_spec is not None:
|
|
new_unfilled.append((step_index, unfilled_spec))
|
|
elif stage_spec is not None:
|
|
new_unfilled.append((step_index, stage_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
|
|
|
|
|
|
def prune_stripped_after_rematch(
|
|
stripped_off_topic: Sequence[Mapping[str, Any]],
|
|
rematch_log: Sequence[Mapping[str, Any]],
|
|
) -> List[Dict[str, Any]]:
|
|
"""Entfernt aus stripped_off_topic Slots, die per Rematch ersetzt wurden."""
|
|
replaced: Set[int] = set()
|
|
for entry in rematch_log or []:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
if str(entry.get("action") or "") != "replaced":
|
|
continue
|
|
midx = entry.get("roadmap_major_step_index")
|
|
if midx is not None:
|
|
replaced.add(int(midx))
|
|
if not replaced:
|
|
return list(stripped_off_topic or [])
|
|
out: List[Dict[str, Any]] = []
|
|
for item in stripped_off_topic or []:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
midx = item.get("roadmap_major_step_index")
|
|
if midx is not None and int(midx) in replaced:
|
|
continue
|
|
out.append(dict(item))
|
|
return out
|
|
|
|
|
|
__all__ = [
|
|
"collect_rematch_slot_indices",
|
|
"prune_stripped_after_rematch",
|
|
"rematch_roadmap_slots",
|
|
]
|