old_state = schedulable.state
            if not schedulable.are_dependencies_met(session=session, dep_context=dep_context):
                old_states[schedulable.key] = old_state
                continue
            
            
            
            
            
            
            new_tis = None
            if schedulable.map_index < 0:
                new_tis = _expand_mapped_task_if_needed(schedulable)
                if new_tis is not None:
                    additional_tis.extend(new_tis)
                    expansion_happened = True
            if new_tis is None and schedulable.state in SCHEDULEABLE_STATES:
                
                
                if schedulable.task.task_id not in revised_map_index_task_ids:
                    ready_tis.extend(self._revise_map_indexes_if_mapped(schedulable.task, session=session))
                    revised_map_index_task_ids.add(schedulable.task.task_id)
                ready_tis.append(schedulable)
        
        tis_filter = TI.filter_for_tis(old_states)
        if tis_filter is not None:
            fresh_tis = session.query(TI).filter(tis_filter).all()
            changed_tis = any(ti.state != old_states[ti.key] for ti in fresh_tis)
        return ready_tis, changed_tis, expansion_happened
    def _are_premature_tis(