(
profile: Dict[str, Any],
candidate: Dict[str, Any],
date: str,
*,
signal_topics: List[str],
settings: Dict[str, Any],
)
| 530 | |
| 531 | def _lock_anchor( |
| 532 | profile: Dict[str, Any], |
| 533 | candidate: Dict[str, Any], |
| 534 | date: str, |
| 535 | *, |
| 536 | signal_topics: List[str], |
| 537 | settings: Dict[str, Any], |
| 538 | ) -> Tuple[Dict[str, Any], Dict[str, Any]]: |
| 539 | updated = copy.deepcopy(profile) |
| 540 | drift_state = _ensure_anchor_state(updated) |
| 541 | anchor_topic = str(drift_state.get("hidden_anchor") or candidate.get("topic") or "").strip() |
| 542 | |
| 543 | drift_state["anchor_topic"] = anchor_topic |
| 544 | drift_state["anchor_topics"] = list(signal_topics or [anchor_topic]) |
| 545 | drift_state["anchor_source"] = "hidden_anchor" |
| 546 | drift_state["anchor_confidence"] = round(float(candidate.get("confidence") or 0.0), 3) |
| 547 | drift_state["anchor_progress"] = 0.0 |
| 548 | drift_state["anchor_set_date"] = date |
| 549 | drift_state["commitment_days_remaining"] = int(settings["commitment_days"]) |
| 550 | drift_state["intent_score"] = round(max(float(drift_state.get("intent_score") or 0.0), float(settings["lock_threshold"])), 2) |
| 551 | drift_state["last_drift_date"] = date |
| 552 | drift_state["baseline_core_directions"] = copy.deepcopy(updated.get("core_directions", {}) or {}) |
| 553 | drift_state["recovery_mode"] = None |
| 554 | drift_state["completed_drift_cycles"] = int(drift_state.get("completed_drift_cycles", 0) or 0) + 1 |
| 555 | if settings.get("max_drift_cycles") is not None and drift_state.get("max_drift_cycles") is None: |
| 556 | drift_state["max_drift_cycles"] = int(settings["max_drift_cycles"]) |
| 557 | updated["drift_state"] = drift_state |
| 558 | |
| 559 | return updated, { |
| 560 | "event_type": "drift", |
| 561 | "method": "anchor_lock", |
| 562 | "anchor_topic": anchor_topic, |
| 563 | "anchor_topics": list(signal_topics or [anchor_topic]), |
| 564 | "anchor_confidence": drift_state["anchor_confidence"], |
| 565 | "commitment_days": int(settings["commitment_days"]), |
| 566 | "completed_drift_cycles": drift_state["completed_drift_cycles"], |
| 567 | } |
| 568 | |
| 569 | |
| 570 | def _advance_anchor_profile( |
| 571 | profile: Dict[str, Any], |
no test coverage detected