Function get_recent_drift_updates

skills/storage-helper/scripts/db_ops.py:1667–1714

Return recent drift-update behavior logs for a user. Args: user_id: User ID days: Lookback days Returns: Parsed drift update records ordered from old to new.

(user_id: str, days: int = 7)

def get_recent_drift_updates(user_id: str, days: int = 7) -> List[Dict[str, Any]]:
    """
    Return recent drift-update behavior logs for a user.

    Args:
        user_id: User ID
        days: Lookback days

    Returns:
        Parsed drift update records ordered from old to new.
    """
    since_timestamp = (datetime.now() - timedelta(days=max(1, int(days)))).isoformat(sep=" ")

    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """
        SELECT id, user_id, push_id, paper_id, action, action_type, category, timestamp, metadata
        FROM behavior_logs
        WHERE user_id = ?
          AND action = 'profile_updated'
          AND action_type = 'drift_update'
          AND timestamp >= ?
        ORDER BY timestamp ASC, id ASC
        """,
        (user_id, since_timestamp),
    )
    rows = cursor.fetchall()
    conn.close()

    results: List[Dict[str, Any]] = []
    for row in rows:
        record = dict(row)
        metadata = _load_json_metadata(record.get("metadata"))
        record["metadata"] = metadata
        if "drift_status" in metadata:
            record["drift_status"] = metadata.get("drift_status")
        if "drift_score" in metadata:
            record["drift_score"] = metadata.get("drift_score")
        if "adaptive_alpha" in metadata:
            record["adaptive_alpha"] = metadata.get("adaptive_alpha")
        if "top_shift_topics" in metadata:
            record["top_shift_topics"] = metadata.get("top_shift_topics")
        if "explanation" in metadata:
            record["explanation"] = metadata.get("explanation")
        results.append(record)

    return results


def get_recent_reading_signal(