async def move_task_confirm(self, *, key, state, stimulus_id, worker=None):
    try:
        ts = self.scheduler.tasks[key]
    except KeyError:
        logger.debug('Key released between request and confirm: %s', key)
        return
    try:
        d = self.in_flight.pop(ts)
        if (d['stimulus_id'] != stimulus_id):
            self.log(('stale-response', key, state, worker, stimulus_id))
            self.in_flight[ts] = d
            return
    except KeyError:
        self.log(('already-aborted', key, state, stimulus_id))
        return
    thief = d['thief']
    victim = d['victim']
    logger.debug('Confirm move %s, %s -> %s.  State: %s', key, victim, thief, state)
    self.in_flight_occupancy[thief] -= d['thief_duration']
    self.in_flight_occupancy[victim] += d['victim_duration']
    if (not self.in_flight):
        self.in_flight_occupancy.clear()
        self._in_flight_event.set()
    if self.scheduler.validate:
        assert (ts.processing_on == victim)
    try:
        _log_msg = [key, state, victim.address, thief.address, stimulus_id]
        if (ts.state != 'processing'):
            self.log(('not-processing', *_log_msg))
            old_thief = thief.occupancy
            new_thief = sum(thief.processing.values())
            old_victim = victim.occupancy
            new_victim = sum(victim.processing.values())
            thief.occupancy = new_thief
            victim.occupancy = new_victim
            self.scheduler.total_occupancy += (((new_thief - old_thief) + new_victim) - old_victim)
        elif ((state in _WORKER_STATE_UNDEFINED) or ((state in _WORKER_STATE_CONFIRM) and (thief.address not in self.scheduler.workers))):
            self.log(('reschedule', (thief.address not in self.scheduler.workers), *_log_msg))
            self.scheduler.reschedule(key)
        elif (state in _WORKER_STATE_REJECT):
            self.log(('already-computing', *_log_msg))
        elif (state in _WORKER_STATE_CONFIRM):
            self.remove_key_from_stealable(ts)
            ts.processing_on = thief
            duration = victim.processing.pop(ts)
            victim.occupancy -= duration
            self.scheduler.total_occupancy -= duration
            if (not victim.processing):
                self.scheduler.total_occupancy -= victim.occupancy
                victim.occupancy = 0
            thief.processing[ts] = d['thief_duration']
            thief.occupancy += d['thief_duration']
            self.scheduler.total_occupancy += d['thief_duration']
            self.put_key_in_stealable(ts)
            self.scheduler.send_task_to_worker(thief.address, ts)
            self.log(('confirm', *_log_msg))
        else:
            raise ValueError(f'Unexpected task state: {state}')
    except Exception as e:
        logger.exception(e)
        if LOG_PDB:
            import pdb
            pdb.set_trace()
        raise
    finally:
        self.scheduler.check_idle_saturated(thief)
        self.scheduler.check_idle_saturated(victim)
async def move_task_confirm(self, *, key, state, stimulus_id, worker=None):
    try:
        ts = self.scheduler.tasks[key]
    except KeyError:
        logger.debug('Key released between request and confirm: %s', key)
        return
    try:
        d = self.in_flight.pop(ts)
        if (d['stimulus_id'] != stimulus_id):
            self.log(('stale-response', key, state, worker, stimulus_id))
            self.in_flight[ts] = d
            return
    except KeyError:
        self.log(('already-aborted', key, state, stimulus_id))
        return
    thief = d['thief']
    victim = d['victim']
    logger.debug('Confirm move %s, %s -> %s.  State: %s', key, victim, thief, state)
    self.in_flight_occupancy[thief] -= d['thief_duration']
    self.in_flight_occupancy[victim] += d['victim_duration']
    if (not self.in_flight):
        self.in_flight_occupancy.clear()
        self._in_flight_event.set()
    if self.scheduler.validate:
        assert (ts.processing_on == victim)
    try:
        _log_msg = [key, state, victim.address, thief.address, stimulus_id]
        if (ts.state != 'processing'):
            self.log(('not-processing', *_log_msg))
            old_thief = thief.occupancy
            new_thief = sum(thief.processing.values())
            old_victim = victim.occupancy
            new_victim = sum(victim.processing.values())
            thief.occupancy = new_thief
            victim.occupancy = new_victim
            self.scheduler.total_occupancy += (((new_thief - old_thief) + new_victim) - old_victim)
        elif ((state in _WORKER_STATE_UNDEFINED) or ((state in _WORKER_STATE_CONFIRM) and (thief.address not in self.scheduler.workers))):
            self.log(('reschedule', (thief.address not in self.scheduler.workers), *_log_msg))
            self.scheduler.reschedule(key)
        elif (state in _WORKER_STATE_REJECT):
            self.log(('already-computing', *_log_msg))
        elif (state in _WORKER_STATE_CONFIRM):
            self.remove_key_from_stealable(ts)
            ts.processing_on = thief
            duration = victim.processing.pop(ts)
            victim.occupancy -= duration
            self.scheduler.total_occupancy -= duration
            if (not victim.processing):
                self.scheduler.total_occupancy -= victim.occupancy
                victim.occupancy = 0
            thief.processing[ts] = d['thief_duration']
            thief.occupancy += d['thief_duration']
            self.scheduler.total_occupancy += d['thief_duration']
            self.put_key_in_stealable(ts)
            self.scheduler.send_task_to_worker(thief.address, ts)
            self.log(('confirm', *_log_msg))
        else:
            raise ValueError(f'Unexpected task state: {state}')
    except Exception as e:
        logger.exception(e)
        if LOG_PDB:
            import pdb
            pdb.set_trace()
        raise
    finally:
        self.scheduler.check_idle_saturated(thief)
        self.scheduler.check_idle_saturated(victim)