Evidence-driven anchored drift: 1. Observe stable multi-day signal for the same new topic 2. Lock an anchor topic once intent is strong enough 3. Progress deterministically for a short commitment window
(
profile: Dict[str, Any],
selected_papers: Optional[List[Dict]],
date: Optional[str],
*,
drift_probability: float = 0.5,
checkfiles: Optional[List[Dict[str, Any]]] = None,
strategy_mode: str = STRATEGY_SIMULATION,
)
| 750 | |
| 751 | def advance_anchor_drift( |
| 752 | profile: Dict[str, Any], |
| 753 | selected_papers: Optional[List[Dict]], |
| 754 | date: Optional[str], |
| 755 | *, |
| 756 | drift_probability: float = 0.5, |
| 757 | checkfiles: Optional[List[Dict[str, Any]]] = None, |
| 758 | strategy_mode: str = STRATEGY_SIMULATION, |
| 759 | ) -> Tuple[Dict[str, Any], Optional[Dict[str, Any]]]: |
| 760 | """ |
| 761 | Evidence-driven anchored drift: |
| 762 | 1. Observe stable multi-day signal for the same new topic |
| 763 | 2. Lock an anchor topic once intent is strong enough |
| 764 | 3. Progress deterministically for a short commitment window |
| 765 | """ |
| 766 | updated = initialize_hidden_anchor( |
| 767 | profile, |
| 768 | strategy_mode=strategy_mode, |
| 769 | drift_probability=drift_probability, |
| 770 | ) |
| 771 | drift_state = _ensure_anchor_state(updated) |
| 772 | settings = _strategy_settings(strategy_mode) |
| 773 | drift_state["strategy_mode"] = settings["mode"] |
| 774 | drift_state["episode_index"] = int(drift_state.get("episode_index") or 0) + 1 |
| 775 | if settings.get("max_drift_cycles") is not None and drift_state.get("max_drift_cycles") is None: |
| 776 | drift_state["max_drift_cycles"] = int(settings["max_drift_cycles"]) |
| 777 | updated["drift_state"] = drift_state |
| 778 | drift_plan = updated.get("drift_plan", {}) or {} |
| 779 | raw_shift_episode_start = max(0, int(drift_plan.get("shift_episode_start", 0) or 0)) |
| 780 | shift_episode_start = min(raw_shift_episode_start, int(settings["max_shift_episode_start"])) if raw_shift_episode_start else 0 |
| 781 | |
| 782 | if settings["mode"] == STRATEGY_SIMULATION and str(drift_state.get("status") or "") == "observing": |
| 783 | signal = _extract_anchor_signal(updated, selected_papers) |
| 784 | drift_state = _append_signal_window(drift_state, date=str(date or ""), signal=signal) |
| 785 | updated["drift_state"] = drift_state |
| 786 | |
| 787 | hidden_anchor_hits = int(signal.get("hidden_anchor_hits") or 0) |
| 788 | if hidden_anchor_hits > 0: |
| 789 | drift_state["intent_score"] = round( |
| 790 | min(1.0, float(drift_state.get("intent_score") or 0.0) + float(settings["intent_increment"])), |
| 791 | 2, |
| 792 | ) |
| 793 | updated["drift_state"] = drift_state |
| 794 | locked_profile, lock_event = _lock_anchor( |
| 795 | updated, |
| 796 | { |
| 797 | "topic": str(drift_state.get("hidden_anchor") or ""), |
| 798 | "confidence": max( |
| 799 | float(signal.get("hidden_anchor_ratio") or 0.0), |
| 800 | float(signal.get("confidence") or 0.0), |
| 801 | ), |
| 802 | }, |
| 803 | str(date or ""), |
| 804 | signal_topics=list(signal.get("source_topics") or []), |
| 805 | settings=settings, |
| 806 | ) |
| 807 | progressed_profile, completion_event = _advance_anchor_profile(locked_profile, str(date or ""), settings=settings) |
| 808 | return progressed_profile, completion_event or lock_event |
| 809 |
no test coverage detected